stream: speed up WHATWG web streams

Avoid per-chunk async wrappers for sync pull/write/start and
complete pipeTo writes without one microtask per chunk. Add a
native webstreams binding with a Fast API isNonThenable check
on the data plane and a memcpy clone for byte views.

Empty stream construction skips redundant validation and lazily
creates the writable AbortController, materializing it on abort()
so controller.signal still reflects the abort reason. Use the
shared kResolvedPromise on the pull/write hot path instead of
allocating PromiseResolve().

Assisted-by: Grok
Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>

Author: anonrig

lib/internal/webstreams/readablestream.js

@@ -111,8 +111,10 @@ const {
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  isNonThenable,
   kEmptyQueue,
   kResolvedPromise,
+  promiseFromAlgorithmResult,
   kState,
   kType,
   lazyTransfer,
@@ -252,6 +254,16 @@ class ReadableStream {
    */
   constructor(source = kEmptyObject, strategy = kEmptyObject) {
     markTransferMode(this, false, true);
+    // Empty-argument `new ReadableStream()`: no source, no strategy, and
+    // no controller. Reads never deliver data, so skip those allocations
+    // until getReader/cancel/error first need a default controller.
+    // Subclasses that call those methods after super() materialize the
+    // controller in the subclass constructor; see
+    // ensureEmptyDefaultController.
+    if (source === kEmptyObject && strategy === kEmptyObject) {
+      this[kState] = createReadableStreamState();
+      return;
+    }
     validateObject(source, 'source', kValidateObjectAllowObjects);
     validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);
     this[kState] = createReadableStreamState();
@@ -301,8 +313,13 @@ class ReadableStream {
     // only default controllers were wired here; byte stream controllers
     // keep the previous no-op behavior.
     const controller = this[kState].controller;
+    if (controller === undefined) {
+      if (this[kState].state === 'readable')
+        readableStreamError(this, error);
+      return;
+    }
     if (isReadableStreamDefaultController(controller))
-      controller.error(error);
+      readableStreamDefaultControllerError(controller, error);
   }
 
   // Used by the internal stream interop (end-of-stream). Materialized
@@ -351,6 +368,7 @@ class ReadableStream {
       return PromiseReject(
         new ERR_INVALID_STATE.TypeError('ReadableStream is locked'));
     }
+    ensureEmptyDefaultController(this);
     return readableStreamCancel(this, reason);
   }
 
@@ -2580,6 +2598,7 @@ function setupReadableStreamBYOBReader(reader, stream) {
 function setupReadableStreamDefaultReader(reader, stream) {
   if (isReadableStreamLocked(stream))
     throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
+  ensureEmptyDefaultController(stream);
   readableStreamReaderGenericInitialize(reader, stream);
   reader[kState].readRequests = kEmptyQueue;
 }
@@ -2729,7 +2748,8 @@ function readableStreamDefaultControllerPull(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.
+  // handles both. Non-thenable results react on kResolvedPromise so each
+  // pull is still separated by a microtask, matching the spec.
   let result;
   try {
     result = controller[kState].pullAlgorithm(controller);
@@ -2763,7 +2783,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableStreamDefaultControllerClearAlgorithms(controller);
-  return result;
+  return promiseFromAlgorithmResult(result);
 }
 
 function readableStreamDefaultControllerPullSteps(controller, readRequest) {
@@ -2796,6 +2816,43 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
     readableStreamDefaultControllerPull(controller);
 }
 
+// Materialize the deferred default controller for `new ReadableStream()`.
+//
+// started is true immediately: the empty-argument start algorithm is a
+// no-op, so there is no initial pull and nothing can observe an unstarted
+// controller without first calling getReader/cancel/pipeTo/tee/values,
+// all of which come through here. That is also why this still matches
+// WPT: those tests either pass a source (leaving this path) or wait for
+// start, which is already complete for a no-op start.
+//
+// Subclasses that call cancel(), getReader(), pipeTo(), tee(), or
+// values() in the constructor body after super() will materialize the
+// controller before the subclass constructor finishes. Passing a source
+// (for example to install start/pull) leaves the empty-argument path
+// and creates the controller during super() as usual.
+function ensureEmptyDefaultController(stream) {
+  if (stream[kState].controller !== undefined)
+    return stream[kState].controller;
+  const controller = new ReadableStreamDefaultController(kSkipThrow);
+  controller[kState] = {
+    cancelAlgorithm: nonOpCancel,
+    closeRequested: false,
+    highWaterMark: 1,
+    pullAgain: false,
+    pullAlgorithm: nonOpCallback,
+    pulling: false,
+    pullFulfilled: undefined,
+    pullRejected: undefined,
+    queue: kEmptyQueue,
+    queueTotalSize: 0,
+    started: true,
+    sizeAlgorithm: defaultSizeAlgorithm,
+    stream,
+  };
+  stream[kState].controller = controller;
+  return controller;
+}
+
 function setupReadableStreamDefaultController(
   stream,
   controller,
@@ -2824,8 +2881,7 @@ function setupReadableStreamDefaultController(
 
   const startResult = startAlgorithm();
 
-  if (startResult === null ||
-      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+  if (isNonThenable(startResult)) {
     // Non-thenable start result: fulfillment is guaranteed and no .then
     // lookup on the result is observable, so run the post-start step
     // directly at the exact microtask position the promise reaction
@@ -3586,7 +3642,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableByteStreamControllerClearAlgorithms(controller);
-  return result;
+  return promiseFromAlgorithmResult(result);
 }
 
 // Dequeues the first chunk of the byte queue as a Uint8Array view,
@@ -3708,8 +3764,7 @@ function setupReadableByteStreamController(
 
   const startResult = startAlgorithm();
 
-  if (startResult === null ||
-      (typeof startResult !== 'object' && typeof startResult !== 'function')) {
+  if (isNonThenable(startResult)) {
     // See setupReadableStreamDefaultController.
     queueMicrotask(() => {
       controller[kState].started = true;

lib/internal/webstreams/transformstream.js

@@ -58,6 +58,7 @@ const {
   kType,
   nonOpCancel,
   nonOpFlush,
+  delayedAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -127,9 +128,14 @@ class TransformStream {
     writableStrategy = kEmptyObject,
     readableStrategy = kEmptyObject) {
     markTransferMode(this, false, true);
-    validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
-    validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
-    validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
+    if (transformer !== kEmptyObject)
+      validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
+    if (writableStrategy !== kEmptyObject) {
+      validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
+    }
+    if (readableStrategy !== kEmptyObject) {
+      validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull);
+    }
     const readableType = transformer?.readableType;
     const writableType = transformer?.writableType;
     const start = transformer?.start;
@@ -648,15 +654,16 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise =
+    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(
     cancelPromise,
     () => {
-      if (readable[kState].state === 'errored')
+      if (readable[kState].state === 'errored') {
         reject(readable[kState].storedError);
-      else {
+      } else {
         readableStreamDefaultControllerError(readable[kState].controller, reason);
         resolve();
       }
@@ -681,7 +688,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
   }
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const flushPromise = controller[kState].flushAlgorithm(controller);
+  const flushPromise =
+    delayedAlgorithmResult(controller[kState].flushAlgorithm(controller));
   transformStreamDefaultControllerClearAlgorithms(controller);
   PromisePrototypeThen(
     flushPromise,
@@ -724,15 +732,16 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise = controller[kState].cancelAlgorithm(reason);
+  const cancelPromise =
+    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
   transformStreamDefaultControllerClearAlgorithms(controller);
 
   PromisePrototypeThen(
     cancelPromise,
     () => {
-      if (writable[kState].state === 'errored')
+      if (writable[kState].state === 'errored') {
         reject(writable[kState].storedError);
-      else {
+      } else {
         writableStreamDefaultControllerErrorIfNeeded(
           writable[kState].controller,
           reason);