stream: decouple transform backpressure changes

The spec's [[backpressureChangePromise]] is only ever observed by the
transform source pull algorithm (settles when backpressure next becomes
true) and by a sink write arriving under backpressure (settles when it
next becomes false), both internal. Replace the promise record with
direct continuation delivery: a parked pull is completed by enqueueing
the readable controller's pull-fulfilled step on the shared resolved
promise, and a parked write by a cached per-stream continuation that
resolves the sink promise with the perform-transform promise, so
adoption reproduces the previous derived-chain settle depth exactly.
Each delivery lands at the same microtask position as the old record's
reaction.
transformStreamDefaultControllerPerformTransform now mirrors the
reference implementation's promiseCall().then(undefined, rejection
steps) directly instead of running an async wrapper pair per chunk: the
transformer.transform callback is wrapped raw, a non-thenable result
reuses the shared resolved promise, and a synchronous throw is
delivered through a rejected promise, keeping the error timing inside a
reaction.
A pipe-through benchmark is added since the suite had no
transform-throughput row.
pipeThrough passthrough improves by ~8% and a writer-driven transform
loop by ~14%; the pipe-to and read families are unchanged.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65143
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: mcollina

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,

benchmark/webstreams/pipe-through.js

@@ -0,0 +1,38 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TransformStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [5e5],
+kind: ['default','transform'],
+});
+
+asyncfunctionmain({ n, kind }){
+constb=Buffer.alloc(64);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(b);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='default' ?
+newTransformStream() :
+newTransformStream({
+transform(chunk,controller){controller.enqueue(chunk);},
+});
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

lib/internal/webstreams/transformstream.js

@@ -5,6 +5,8 @@ const {
  ObjectDefineProperties,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
+ PromiseReject,
+ PromiseResolve,
  PromiseWithResolvers,
  Symbol,
  SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const{
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
+ kParkedAlgorithmResult,
+ kResolvedPromise,
  kState,
  kType,
  nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
 readable: undefined,
 writable: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 controller: undefined,
 };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 constisTransformStreamDefaultController=
 isBrandCheck('TransformStreamDefaultController');
 
-asyncfunctiondefaultTransformAlgorithm(chunk,controller){
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+functiondefaultTransformAlgorithm(chunk,controller){
 transformStreamDefaultControllerEnqueue(controller,chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
  writable,
 controller: undefined,
 backpressure: undefined,
-backpressureChange: undefined,
+// Continuation slots replacing the spec's
+// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+pullPending: false,
+pendingWrite: undefined,
+pendingWriteChunk: undefined,
+writeContinuation: undefined,
 };
 
 transformStreamSetBackpressure(stream,true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-functiontransformStreamBackpressureChangePromise(stream){
-conststate=stream[kState];
-return(state.backpressureChange??=PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 functiontransformStreamSetBackpressure(stream,backpressure){
 conststate=stream[kState];
 assert(state.backpressure!==backpressure);
-constbackpressureChange=state.backpressureChange;
-if(backpressureChange!==undefined){
-state.backpressureChange=undefined;
-backpressureChange.resolve();
-}
 state.backpressure=backpressure;
+if(backpressure){
+if(state.pullPending){
+state.pullPending=false;
+// The pull-fulfilled step exists: a pull parked it (see
+// transformStreamDefaultSourcePullAlgorithm), and the readable
+// controller creates it before invoking the pull algorithm.
+PromisePrototypeThen(
+kResolvedPromise,
+state.readable[kState].controller[kState].pullFulfilled);
+}
+}elseif(state.pendingWrite!==undefined){
+PromisePrototypeThen(kResolvedPromise,state.writeContinuation);
+}
 }
 
 functionsetupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
  transformAlgorithm,
  flushAlgorithm,
  cancelAlgorithm,
+performTransformRejected: undefined,
 };
 stream[kState].controller=controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
 constflush=transformer?.flush;
 constcancel=transformer?.cancel;
 consttransformAlgorithm=transform ?
-createPromiseCallback2Params('transformer.transform',transform,transformer) :
+createRawCallback2Params('transformer.transform',transform,transformer) :
 defaultTransformAlgorithm;
 constflushAlgorithm=flush ?
 createPromiseCallback1Param('transformer.flush',flush,transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
 transformStreamError(controller[kState].stream,error);
 }
 
-asyncfunctiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+functiontransformStreamDefaultControllerPerformTransform(controller,chunk){
+constcontrollerState=controller[kState];
+consttransformAlgorithm=controllerState.transformAlgorithm;
+if(transformAlgorithm===undefined){
+// Algorithms were cleared by a concurrent cancel/abort/close.
+returnkResolvedPromise;
+}
+letresult;
 try{
-consttransformAlgorithm=controller[kState].transformAlgorithm;
-if(transformAlgorithm===undefined){
-// Algorithms were cleared by a concurrent cancel/abort/close.
-return;
-}
-returnawaittransformAlgorithm(chunk,controller);
+result=transformAlgorithm(chunk,controller);
 }catch(error){
+result=PromiseReject(error);
+}
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+result=kResolvedPromise;
+}else{
+result=PromiseResolve(result);
+}
+controllerState.performTransformRejected??=(error)=>{
 transformStreamError(controller[kState].stream,error);
 throwerror;
-}
+};
+returnPromisePrototypeThen(
+result,
+undefined,
+controllerState.performTransformRejected);
 }
 
 functiontransformStreamDefaultControllerTerminate(controller){
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 functiontransformStreamDefaultSinkWriteAlgorithm(stream,chunk){
+conststate=stream[kState];
 const{
  writable,
  controller,
-}=stream[kState];
+}=state;
 assert(writable[kState].state==='writable');
-if(stream[kState].backpressure){
-constbackpressureChange=transformStreamBackpressureChangePromise(stream);
-returnPromisePrototypeThen(
-backpressureChange,
-()=>{
-const{
- writable,
-}=stream[kState];
-if(writable[kState].state==='erroring')
-throwwritable[kState].storedError;
-assert(writable[kState].state==='writable');
-returntransformStreamDefaultControllerPerformTransform(
+if(state.backpressure){
+// Park the chunk and one promise record; the backpressure -> false
+// flip delivers the cached continuation (see
+// transformStreamSetBackpressure) at the same microtask position as
+// the old [[backpressureChangePromise]] reaction. The continuation
+// resolves the sink promise with the perform-transform promise, so
+// adoption reproduces the old derived-chain settle depth exactly.
+// The writable dispatches a single write at a time, so one pending
+// slot suffices.
+assert(state.pendingWrite===undefined);
+constpendingWrite=PromiseWithResolvers();
+state.pendingWrite=pendingWrite;
+state.pendingWriteChunk=chunk;
+state.writeContinuation??=()=>{
+constpending=state.pendingWrite;
+constpendingChunk=state.pendingWriteChunk;
+state.pendingWrite=undefined;
+state.pendingWriteChunk=undefined;
+constwritableState=state.writable[kState];
+if(writableState.state==='erroring'){
+pending.reject(writableState.storedError);
+return;
+}
+assert(writableState.state==='writable');
+pending.resolve(
+transformStreamDefaultControllerPerformTransform(
 controller,
-chunk);
-});
+pendingChunk));
+};
+returnpendingWrite.promise;
 }
 returntransformStreamDefaultControllerPerformTransform(controller,chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 functiontransformStreamDefaultSourcePullAlgorithm(stream){
-assert(stream[kState].backpressure);
+conststate=stream[kState];
+assert(state.backpressure);
 transformStreamSetBackpressure(stream,false);
-returntransformStreamBackpressureChangePromise(stream);
+// Park the pull: the next backpressure -> true flip delivers the
+// pull-fulfilled step (see transformStreamSetBackpressure). The old
+// [[backpressureChangePromise]] this replaces was only ever resolved,
+// so the parked pull needs no rejection delivery.
+state.pullPending=true;
+returnkParkedAlgorithmResult;
 }
 
 functiontransformStreamDefaultSourceCancelAlgorithm(stream,reason){

lib/internal/webstreams/util.js

@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 constkResolvedPromise=PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+constkParkedAlgorithmResult={__proto__: null};
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===kParkedAlgorithmResult)
+return;
 if(result===null||
 (typeofresult!=='object'&&typeofresult!=='function')){
 PromisePrototypeThen(kResolvedPromise,onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kParkedAlgorithmResult,
  kResolvedPromise,
  kState,
  kType,