stream: cut promise churn in webstreams hot paths

Three related reductions on the per-chunk paths:
Wrap user sink.write and source.pull callbacks without coercing their
result into a promise. When the callback returns a non-thenable (the
common synchronous case), fulfillment is guaranteed and no then() lookup
is observable, so the fulfilled reaction is enqueued through a single
shared resolved promise at the exact microtask position the coerced
promise's reaction would have had, skipping the implicit async-wrapper
promise per chunk. Thenable results go through PromiseResolve(), which
matches the spec's "a promise resolved with" conversion (identity for
native promises).
Park pipeTo's pump on backpressure by installing a record that
duck-types the writer's lazily-materialized [[readyPromise]] record and
whose resolve function is the pump continuation itself. Backpressure
clearing then resumes the pump directly instead of materializing a fresh
promise record plus reaction per flip, and the pump no longer schedules
a microtask per batch. writableStreamUpdateBackpressure publishes the
new backpressure state before resolving the ready record so the pump
observes the updated value.
Replace queueMicrotask() on the pipeTo and tee chunk-forwarding paths
with a reaction on the shared resolved promise, which enqueues the
continuation at the same position without the per-call scheduling
overhead.
pipe-to improves by 8-14% across all benchmark configurations, with
readable-read and tee also improving in spot runs.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65138
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

Author: mcollina

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallback1Param,
+ createRawCallback1Param,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -110,6 +111,7 @@ const {
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -121,6 +123,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -137,7 +140,6 @@ const {
  writableStreamDefaultWriterRelease,
  writableStreamDefaultWriterWriteWithRequest,
  writerClosedPromise,
- writerReadyPromise,
 }=require('internal/webstreams/writablestream');
 
 const{ Buffer }=require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
 // the chunk travels through `pendingChunk`.
 letpendingChunk;
 letreadRequest;
+letreadyHook;
 
 // Ready promise rejection is handled by the destination-errored
 // watcher.
 functionignoreReadyRejection(){}
 
+// Parks the pump on the destination's backpressure by installing a
+// record that duck-types the writer's lazily-materialized
+// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+// when backpressure clears (after publishing the new backpressure
+// state), which re-enters the pump directly instead of rotating a
+// fresh promise record plus reaction per flip. The pipe holds the only
+// reference to the writer, so the record is never observable as a real
+// ready promise; the erroring/release paths probe `promise` via
+// isPromisePending() and call `reject`, so it carries a real
+// forever-pending promise and a no-op reject.
+functionparkOnReady(){
+readyHook??={
+promise: newPromise(nonOpCallback),
+resolve: pump,
+reject: ignoreReadyRejection,
+};
+writer[kState].ready=readyHook;
+}
+
 functionforwardChunk(){
 constchunk=pendingChunk;
 pendingChunk=undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
 if(shuttingDown)return;
 
 if(dest[kState].backpressure){
-PromisePrototypeThen(
-writerReadyPromise(writer).promise,
-pump,
-ignoreReadyRejection);
+parkOnReady();
 return;
 }
 
@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
 return;
 }
 
-// Yield to microtask queue between batches to allow events/signals
-// to fire
-queueMicrotask(pump);
+// Park on backpressure directly: the ready hook resumes the pump
+// when a completed write clears it.
+if(dest[kState].backpressure){
+parkOnReady();
+return;
+}
+
+// Yield to the microtask queue between batches so completed-write
+// reactions and events/signals fire; a shared resolved promise
+// enqueues the continuation at the same position as queueMicrotask
+// without the per-batch scheduling overhead.
+PromisePrototypeThen(kResolvedPromise,pump);
 return;
 }
 
@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
 // synchronous write during enqueue(). See WHATWG Streams spec
 // "ReadableStreamPipeTo" step 15's "chunk steps".
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){},
 [kError](){},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
 // The microtask is required by the spec (ReadableStreamTee's
 // "chunk steps" queue one).
 pendingChunk=value;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 // The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
 defaultReadRequest??={
 [kChunk](chunk){
 pendingChunk=chunk;
-queueMicrotask(forwardChunk);
+PromisePrototypeThen(kResolvedPromise,forwardChunk);
 },
 [kClose](){
 reading=false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
 controller[kState].pullRejected=
 (error)=>readableStreamDefaultControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// The pull algorithm may be a raw callback (a wrapped user source.pull
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
 controller[kState].pullRejected=
 (error)=>readableByteStreamControllerError(controller,error);
 }
-PromisePrototypeThen(
-controller[kState].pullAlgorithm(controller),
+// See readableStreamDefaultControllerPull for the raw-callback contract.
+letresult;
+try{
+result=controller[kState].pullAlgorithm(controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].pullFulfilled,
 controller[kState].pullRejected);
 }
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
 FunctionPrototypeBind(start,source,controller) :
 nonOpStart;
 constpullAlgorithm=pull ?
-createPromiseCallback1Param('source.pull',pull,source) :
+createRawCallback1Param('source.pull',pull,source) :
 nonOpPull;
 constcancelAlgorithm=cancel ?
 createPromiseCallback1Param('source.cancel',cancel,source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
 returnasync()=>FunctionPrototypeCall(fn,thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+functioncreateRawCallback1Param(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
+}
+
+functioncreateRawCallback2Params(name,fn,thisArg){
+validateFunction(fn,name);
+return(arg1,arg2)=>FunctionPrototypeCall(fn,thisArg,arg1,arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+constkResolvedPromise=PromiseResolve();
+
+// 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
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+functionthenAlgorithmResult(result,onFulfilled,onRejected){
+if(result===null||
+(typeofresult!=='object'&&typeofresult!=='function')){
+PromisePrototypeThen(kResolvedPromise,onFulfilled);
+}else{
+PromisePrototypeThen(PromiseResolve(result),onFulfilled,onRejected);
+}
+}
+
 functioncreatePromiseCallback1Param(name,fn,thisArg){
 validateFunction(fn,name);
 returnasync(arg)=>FunctionPrototypeCall(fn,thisArg,arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 functionnonOpStart(){}
 
-asyncfunctionnonOpPull(){}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+functionnonOpPull(){}
 
 asyncfunctionnonOpCancel(){}
 
-asyncfunctionnonOpWrite(){}
+functionnonOpWrite(){}
 
 lettransfer;
 functionlazyTransfer(){
@@ -411,6 +448,8 @@ module.exports = {
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
+ createRawCallback1Param,
+ createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
  isBrandCheck,
  isPromisePending,
  kEmptyQueue,
+ kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
  Queue,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
-createPromiseCallback2Params,
+createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  dequeueValue,
@@ -78,6 +78,7 @@ const {
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
+ thenAlgorithmResult,
 }=require('internal/webstreams/util');
 
 const{
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 constbackpressure=
 controllerState.highWaterMark-controllerState.queueTotalSize<=0;
 constwriter=streamState.writer;
-if(writer!==undefined&&streamState.backpressure!==backpressure){
+constchanged=streamState.backpressure!==backpressure;
+// The state field is published before the ready record is resolved so
+// that a ready resolve hook (pipeTo's pump continuation) observes the
+// new value.
+streamState.backpressure=backpressure;
+if(writer!==undefined&&changed){
 if(backpressure){
 // The spec replaces [[readyPromise]] with a fresh pending promise;
 // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
 writer[kState].ready?.resolve();
 }
 }
-streamState.backpressure=backpressure;
 }
 
 functionwritableStreamStartErroring(stream,reason){
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 };
 }
 
-PromisePrototypeThen(
-writeAlgorithm(chunk,controller),
+// The write algorithm may be a raw callback (a wrapped user sink.write
+// returns its result uncoerced; a synchronous throw surfaces here) or an
+// internal algorithm that always returns a promise; thenAlgorithmResult
+// handles both.
+letresult;
+try{
+result=writeAlgorithm(chunk,controller);
+}catch(error){
+result=PromiseReject(error);
+}
+thenAlgorithmResult(
+result,
 controller[kState].writeFulfilled,
 controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
 FunctionPrototypeBind(start,sink,controller) :
 nonOpStart;
 constwriteAlgorithm=write ?
-createPromiseCallback2Params('sink.write',write,sink) :
+createRawCallback2Params('sink.write',write,sink) :
 nonOpWrite;
 constcloseAlgorithm=close ?
 createPromiseCallbackNoParams('sink.close',close,sink) :