stream: only pass the expected number of parameters to callbacks

lib/internal/webstreams/readablestream.js

@@ -104,7 +104,7 @@ const {
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
-  createPromiseCallback,
+  createPromiseCallback1Param,
   customInspect,
   dequeueValue,
   enqueueValueWithSize,
@@ -2582,10 +2582,10 @@ function setupReadableStreamDefaultControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback('source.pull', pull, source) :
+    createPromiseCallback1Param('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('source.cancel', cancel, source) :
+    createPromiseCallback1Param('source.cancel', cancel, source) :
     nonOpCancel;
 
   setupReadableStreamDefaultController(
@@ -3405,10 +3405,10 @@ function setupReadableByteStreamControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback('source.pull', pull, source, controller) :
+    createPromiseCallback1Param('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('source.cancel', cancel, source) :
+    createPromiseCallback1Param('source.cancel', cancel, source) :
     nonOpCancel;
 
   if (autoAllocateChunkSize === 0) {

lib/internal/webstreams/transformstream.js

@@ -43,7 +43,8 @@ const {
 } = require('internal/worker/js_transferable');
 
 const {
-  createPromiseCallback,
+  createPromiseCallback1Param,
+  createPromiseCallback2Params,
   customInspect,
   extractHighWaterMark,
   extractSizeAlgorithm,
@@ -462,13 +463,13 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const flush = transformer?.flush;
   const cancel = transformer?.cancel;
   const transformAlgorithm = transform ?
-    createPromiseCallback('transformer.transform', transform, transformer) :
+    createPromiseCallback2Params('transformer.transform', transform, transformer) :
     defaultTransformAlgorithm;
   const flushAlgorithm = flush ?
-    createPromiseCallback('transformer.flush', flush, transformer) :
+    createPromiseCallback1Param('transformer.flush', flush, transformer) :
     nonOpFlush;
   const cancelAlgorithm = cancel ?
-    createPromiseCallback('transformer.cancel', cancel, transformer) :
+    createPromiseCallback1Param('transformer.cancel', cancel, transformer) :
     nonOpCancel;
 
   setupTransformStreamDefaultController(

lib/internal/webstreams/util.js

@@ -7,10 +7,10 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeShift,
   AsyncIteratorPrototype,
+  FunctionPrototypeCall,
   MathMax,
   NumberIsNaN,
   PromisePrototypeThen,
-  ReflectApply,
   ReflectGet,
   Symbol,
   Uint8Array,
@@ -169,9 +169,25 @@ function enqueueValueWithSize(controller, value, size) {
   controller[kState].queueTotalSize += size;
 }
 
-function createPromiseCallback(name, fn, thisArg) {
+// Arity-specialized variants of the promise-callback wrapper. The generic
+// rest-parameter + ReflectApply form allocated an arguments array on every
+// invocation; these run on per-chunk hot paths (pull/write/transform), so
+// each known call-site arity gets its own wrapper. The exact number of
+// arguments passed through to the user callback is observable and must be
+// preserved.
+function createPromiseCallbackNoParams(name, fn, thisArg) {
   validateFunction(fn, name);
-  return async (...args) => ReflectApply(fn, thisArg, args);
+  return async () => FunctionPrototypeCall(fn, thisArg);
+}
+
+function createPromiseCallback1Param(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
+}
+
+function createPromiseCallback2Params(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
 }
 
 function isPromisePending(promise) {
@@ -213,7 +229,9 @@ module.exports = {
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
-  createPromiseCallback,
+  createPromiseCallbackNoParams,
+  createPromiseCallback1Param,
+  createPromiseCallback2Params,
   customInspect,
   dequeueValue,
   enqueueValueWithSize,

lib/internal/webstreams/writablestream.js

@@ -55,7 +55,9 @@ const {
 } = require('internal/worker/js_transferable');
 
 const {
-  createPromiseCallback,
+  createPromiseCallbackNoParams,
+  createPromiseCallback1Param,
+  createPromiseCallback2Params,
   customInspect,
   dequeueValue,
   enqueueValueWithSize,
@@ -1272,13 +1274,13 @@ function setupWritableStreamDefaultControllerFromSink(
     FunctionPrototypeBind(start, sink, controller) :
     nonOpStart;
   const writeAlgorithm = write ?
-    createPromiseCallback('sink.write', write, sink) :
+    createPromiseCallback2Params('sink.write', write, sink) :
     nonOpWrite;
   const closeAlgorithm = close ?
-    createPromiseCallback('sink.close', close, sink) :
+    createPromiseCallbackNoParams('sink.close', close, sink) :
     nonOpCancel;
   const abortAlgorithm = abort ?
-    createPromiseCallback('sink.abort', abort, sink) :
+    createPromiseCallback1Param('sink.abort', abort, sink) :
     nonOpCancel;
   setupWritableStreamDefaultController(
     stream,