stream: trim webstreams sink wrappers and BYOB transfers

lib/internal/webstreams/readablestream.js

@@ -2388,13 +2388,16 @@ function readableByteStreamControllerConvertPullIntoDescriptor(desc) {
   if (bytesFilled > byteLength)
     throw new ERR_INVALID_STATE.RangeError('The buffer size is invalid');
   assert(!(bytesFilled % elementSize));
-  const transferredBuffer = ArrayBufferPrototypeTransfer(buffer);
+  // The spec transfers the buffer once more here. Nothing else can reach
+  // it by now: it was transferred when the descriptor was created, and
+  // the BYOB request that exposes it has been invalidated, which always
+  // goes together with a transfer, so the view is built on it as-is.
 
   if (ctor === Buffer) {
-    return Buffer.from(transferredBuffer, byteOffset, bytesFilled / elementSize);
+    return Buffer.from(buffer, byteOffset, bytesFilled / elementSize);
   }
 
-  return new ctor(transferredBuffer, byteOffset, bytesFilled / elementSize);
+  return new ctor(buffer, byteOffset, bytesFilled / elementSize);
 }
 
 function isReadableStreamLocked(stream) {
@@ -3427,11 +3430,15 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
       );
     }
 
-    readableByteStreamControllerInvalidateBYOBRequest(controller);
-
-    firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer(
-      firstPendingPullInto.buffer,
-    );
+    // The transfer detaches the view the BYOB request exposed; without
+    // a request the buffer is unreachable and there is nothing to
+    // detach.
+    if (controller[kState].byobRequest !== null) {
+      readableByteStreamControllerInvalidateBYOBRequest(controller);
+      firstPendingPullInto.buffer = ArrayBufferPrototypeTransfer(
+        firstPendingPullInto.buffer,
+      );
+    }
 
     if (firstPendingPullInto.type === 'none') {
       readableByteStreamControllerEnqueueDetachedPullIntoToQueue(

lib/internal/webstreams/transformstream.js

@@ -35,6 +35,7 @@ const {
 } = require('internal/util');
 
 const {
+  validateFunction,
   validateObject,
   kValidateObjectAllowObjects,
   kValidateObjectAllowObjectsAndNull,
@@ -48,7 +49,6 @@ const {
 } = require('internal/worker/js_transferable');
 
 const {
-  createPromiseCallback1Param,
   createRawCallback2Params,
   customInspect,
   deferredStartAlgorithm,
@@ -60,8 +60,8 @@ const {
   kResolvedPromise,
   kState,
   kType,
-  nonOpCancel,
-  nonOpFlush,
+  nonOpCallback,
+  promiseCall1Param,
 } = require('internal/webstreams/util');
 
 const {
@@ -273,6 +273,9 @@ ObjectSetPrototypeOf(TransformStreamState.prototype, null);
 
 class TransformStreamControllerState {
   stream = undefined;
+  // The user-provided transformer the flush and cancel algorithms are
+  // methods of.
+  transformer = undefined;
   transformAlgorithm = undefined;
   flushAlgorithm = undefined;
   cancelAlgorithm = undefined;
@@ -517,11 +520,13 @@ function setupTransformStreamDefaultController(
   controller,
   transformAlgorithm,
   flushAlgorithm,
-  cancelAlgorithm) {
+  cancelAlgorithm,
+  transformer) {
   assert(isTransformStream(stream));
   assert(stream[kState].controller === undefined);
   const controllerState = new TransformStreamControllerState();
   controllerState.stream = stream;
+  controllerState.transformer = transformer;
   controllerState.transformAlgorithm = transformAlgorithm;
   controllerState.flushAlgorithm = flushAlgorithm;
   controllerState.cancelAlgorithm = cancelAlgorithm;
@@ -539,25 +544,25 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const transformAlgorithm = transform ?
     createRawCallback2Params('transformer.transform', transform, transformer) :
     defaultTransformAlgorithm;
-  const flushAlgorithm = flush ?
-    createPromiseCallback1Param('transformer.flush', flush, transformer) :
-    nonOpFlush;
-  const cancelAlgorithm = cancel ?
-    createPromiseCallback1Param('transformer.cancel', cancel, transformer) :
-    nonOpCancel;
+  if (flush)
+    validateFunction(flush, 'transformer.flush');
+  if (cancel)
+    validateFunction(cancel, 'transformer.cancel');
 
   setupTransformStreamDefaultController(
     stream,
     controller,
     transformAlgorithm,
-    flushAlgorithm,
-    cancelAlgorithm);
+    flush || nonOpCallback,
+    cancel || nonOpCallback,
+    transformer);
 }
 
 function transformStreamDefaultControllerClearAlgorithms(controller) {
   controller[kState].transformAlgorithm = undefined;
   controller[kState].flushAlgorithm = undefined;
   controller[kState].cancelAlgorithm = undefined;
+  controller[kState].transformer = undefined;
 }
 
 function transformStreamDefaultControllerEnqueue(controller, chunk) {
@@ -715,7 +720,9 @@ async function transformStreamDefaultSinkAbortAlgorithm(
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise = promiseCall1Param(
+    controller[kState].cancelAlgorithm, controller[kState].transformer,
+    reason);
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(
@@ -749,7 +756,9 @@ function transformStreamDefaultSinkCloseAlgorithm(writableController) {
   }
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const flushPromise = controller[kState].flushAlgorithm(controller);
+  const flushPromise = promiseCall1Param(
+    controller[kState].flushAlgorithm, controller[kState].transformer,
+    controller);
   transformStreamDefaultControllerClearAlgorithms(controller);
   PromisePrototypeThen(
     flushPromise,
@@ -796,7 +805,9 @@ function transformStreamDefaultSourceCancelAlgorithm(
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise = promiseCall1Param(
+    controller[kState].cancelAlgorithm, controller[kState].transformer,
+    reason);
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(

lib/internal/webstreams/util.js

@@ -323,20 +323,14 @@ function enqueueValueWithSize(controller, value, size) {
   state.queueTotalSize += coercedSize;
 }
 
-// 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 () => 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.
+// The per-chunk callbacks of a user-provided source, sink or transformer
+// (pull, write, transform) are wrapped in a closure whose captured method
+// stays a constant for the optimizing compiler. Each known call-site
+// arity gets its own wrapper, since a rest-parameter form would allocate
+// an arguments array per invocation and the exact number of arguments
+// passed to the user callback is observable. Consumers of a raw callback
+// invoke it inside try/catch and route the result through
+// thenAlgorithmResult() below.
 function createRawCallback1Param(name, fn, thisArg) {
   validateFunction(fn, name);
   return (arg) => FunctionPrototypeCall(fn, thisArg, arg);
@@ -347,6 +341,46 @@ function createRawCallback2Params(name, fn, thisArg) {
   return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
 }
 
+// Async wrapper of a one-shot callback (the source cancel method).
+function createPromiseCallback1Param(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
+}
+
+// The one-shot methods of a sink or transformer (close, abort, flush,
+// cancel) are instead stored as-is in the controller state next to the
+// object they belong to and called the way the reference implementation's
+// promiseCall() does: a synchronous throw becomes a rejection and the
+// result is adopted through a fresh promise, so a returned promise settles
+// the caller two microtasks later and a plain value settles it at once.
+function promiseCall1Param(fn, thisArg, arg) {
+  let result;
+  try {
+    result = FunctionPrototypeCall(fn, thisArg, arg);
+  } catch (error) {
+    return PromiseReject(error);
+  }
+  return promiseResolvedWith(result);
+}
+
+function promiseCallNoParams(fn, thisArg) {
+  let result;
+  try {
+    result = FunctionPrototypeCall(fn, thisArg);
+  } catch (error) {
+    return PromiseReject(error);
+  }
+  return promiseResolvedWith(result);
+}
+
+function promiseResolvedWith(value) {
+  if (value === null ||
+      (typeof value !== 'object' && typeof value !== 'function')) {
+    return PromiseResolve(value);
+  }
+  return new Promise((resolve) => resolve(value));
+}
+
 // A single shared, forever-resolved promise used to enqueue a reaction at
 // the next microtask checkpoint without allocating a fresh promise.
 const kResolvedPromise = PromiseResolve();
@@ -385,16 +419,6 @@ function thenAlgorithmResult(result, onFulfilled, onRejected) {
   }
 }
 
-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) {
   if (promise === undefined) return false;
   const details = getPromiseDetails(promise);
@@ -438,12 +462,11 @@ function setPromiseHandled(promise) {
   PromisePrototypeThen(promise, undefined, () => {});
 }
 
-async function nonOpFlush() {}
-
 // Shared non-op for the start/pull/write algorithm callbacks, which all
-// follow the raw-callback contract (see createRawCallback*): the
+// follow the raw-callback contract (see createRawCallback*), and for the
+// missing one-shot methods of a user-provided sink or transformer: the
 // non-thenable return takes the allocation-free fast path in
-// thenAlgorithmResult().
+// thenAlgorithmResult() and settles promiseCall*() at once.
 function nonOpCallback() {}
 
 async function nonOpCancel() {}
@@ -463,9 +486,7 @@ module.exports = {
   canCopyArrayBuffer,
   cloneAsUint8Array,
   copyArrayBuffer,
-  createPromiseCallbackNoParams,
   createPromiseCallback1Param,
-  createPromiseCallback2Params,
   createRawCallback1Param,
   createRawCallback2Params,
   customInspect,
@@ -489,9 +510,9 @@ module.exports = {
   materializeQueue,
   nonOpCallback,
   nonOpCancel,
-  nonOpFlush,
-
   peekQueueValue,
+  promiseCall1Param,
+  promiseCallNoParams,
   rejectedHandledRecord,
   resetQueue,
   resolvedRecord,