stream: address webstreams review blockers

Stop sharing kResolvedPromise on writer.ready/closed and cancel().
Restore async wrappers for cancel/close/abort/flush/transform so
thenable results keep the previous microtask count. Remove the
write-queue drain loop so each write stays one microtask apart.

Drop the native webstreams binding. isNonThenable and
cloneAsUint8Array stay in JS so a detached buffer still throws
TypeError. Initialize the deferred controller field and skip
materializing it on cancel of a non-readable empty stream.

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

Author: anonrig

lib/internal/webstreams/readablestream.js

@@ -114,7 +114,6 @@ const {
   isNonThenable,
   kEmptyQueue,
   kResolvedPromise,
-  promiseFromAlgorithmResult,
   kState,
   kType,
   lazyTransfer,
@@ -314,8 +313,9 @@ class ReadableStream {
     // keep the previous no-op behavior.
     const controller = this[kState].controller;
     if (controller === undefined) {
-      if (this[kState].state === 'readable')
+      if (this[kState].state === 'readable') {
         readableStreamError(this, error);
+      }
       return;
     }
     if (isReadableStreamDefaultController(controller))
@@ -368,7 +368,10 @@ class ReadableStream {
       return PromiseReject(
         new ERR_INVALID_STATE.TypeError('ReadableStream is locked'));
     }
-    ensureEmptyDefaultController(this);
+    // Only materialize the deferred empty controller when cancel will
+    // actually run cancel steps. closed/errored streams return immediately.
+    if (this[kState].state === 'readable')
+      ensureEmptyDefaultController(this);
     return readableStreamCancel(this, reason);
   }
 
@@ -1440,6 +1443,7 @@ function createReadableStreamState() {
   return {
     __proto__: null,
     closedPromise: undefined,
+    controller: undefined,
     disturbed: false,
     reader: undefined,
     state: 'readable',
@@ -2783,7 +2787,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableStreamDefaultControllerClearAlgorithms(controller);
-  return promiseFromAlgorithmResult(result);
+  return result;
 }
 
 function readableStreamDefaultControllerPullSteps(controller, readRequest) {
@@ -3642,7 +3646,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
   resetQueue(controller);
   const result = controller[kState].cancelAlgorithm(reason);
   readableByteStreamControllerClearAlgorithms(controller);
-  return promiseFromAlgorithmResult(result);
+  return result;
 }
 
 // Dequeues the first chunk of the byte queue as a Uint8Array view,

lib/internal/webstreams/transformstream.js

@@ -54,7 +54,6 @@ const {
   kType,
   nonOpCancel,
   nonOpFlush,
-  delayedAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -124,8 +123,9 @@ class TransformStream {
     writableStrategy = kEmptyObject,
     readableStrategy = kEmptyObject) {
     markTransferMode(this, false, true);
-    if (transformer !== kEmptyObject)
+    if (transformer !== kEmptyObject) {
       validateObject(transformer, 'transformer', kValidateObjectAllowObjects);
+    }
     if (writableStrategy !== kEmptyObject) {
       validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull);
     }
@@ -354,7 +354,7 @@ const isTransformStream =
 const isTransformStreamDefaultController =
   isBrandCheck('TransformStreamDefaultController');
 
-function defaultTransformAlgorithm(chunk, controller) {
+async function defaultTransformAlgorithm(chunk, controller) {
   transformStreamDefaultControllerEnqueue(controller, chunk);
 }
 
@@ -595,16 +595,15 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise =
-    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
+  const cancelPromise = 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();
       }
@@ -629,8 +628,7 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
   }
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const flushPromise =
-    delayedAlgorithmResult(controller[kState].flushAlgorithm(controller));
+  const flushPromise = controller[kState].flushAlgorithm(controller);
   transformStreamDefaultControllerClearAlgorithms(controller);
   PromisePrototypeThen(
     flushPromise,
@@ -667,16 +665,15 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
 
   const { promise, resolve, reject } = PromiseWithResolvers();
   controller[kState].finishPromise = promise;
-  const cancelPromise =
-    delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason));
+  const cancelPromise = 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);

lib/internal/webstreams/util.js

@@ -4,6 +4,7 @@ const {
   Array,
   ArrayBufferPrototypeGetByteLength,
   ArrayBufferPrototypeGetDetached,
+  ArrayBufferPrototypeSlice,
   AsyncIteratorPrototype,
   DataViewPrototypeGetBuffer,
   DataViewPrototypeGetByteLength,
@@ -19,6 +20,7 @@ const {
   TypedArrayPrototypeGetBuffer,
   TypedArrayPrototypeGetByteLength,
   TypedArrayPrototypeGetByteOffset,
+  Uint8Array,
 } = primordials;
 
 const {
@@ -31,11 +33,6 @@ const {
   copyArrayBuffer,
 } = internalBinding('buffer');
 
-const {
-  isNonThenable,
-  cloneAsUint8Array: nativeCloneAsUint8Array,
-} = internalBinding('webstreams');
-
 const {
   inspect,
 } = require('util');
@@ -131,7 +128,22 @@ function ArrayBufferViewGetByteOffset(view) {
 }
 
 function cloneAsUint8Array(view) {
-  return nativeCloneAsUint8Array(view);
+  const buffer = ArrayBufferViewGetBuffer(view);
+  const byteOffset = ArrayBufferViewGetByteOffset(view);
+  const byteLength = ArrayBufferViewGetByteLength(view);
+  return new Uint8Array(
+    ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength),
+  );
+}
+
+// True when `value` cannot be a thenable: null, undefined, or a
+// non-object non-function primitive. Objects and functions are treated
+// as maybe-thenable without looking up `.then` (that lookup is
+// observable). Proxies of objects/functions take the maybe-thenable
+// path; a Proxy around a primitive is still an object.
+function isNonThenable(value) {
+  return value === null ||
+    (typeof value !== 'object' && typeof value !== 'function');
 }
 
 function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) {
@@ -332,19 +344,13 @@ function enqueueValueWithSize(controller, value, size) {
 // arguments passed through to the user callback is observable and must be
 // preserved.
 //
-// These are intentionally not `async` functions and not `Promise.try`.
-// Both always allocate a Promise, even when the user callback is
-// synchronous and returns a non-thenable. Callers use `isNonThenable()`
-// (or `PromisePrototypeThen` for thenables) to settle the result.
+// Cold algorithms (cancel/close/abort/flush/transform) stay `async` so
+// a user thenable is adopted with the same microtask count as before.
+// Pull/write use the raw-callback contract instead (see
+// createRawCallback*) and route results through thenAlgorithmResult().
 function createPromiseCallbackNoParams(name, fn, thisArg) {
   validateFunction(fn, name);
-  return () => {
-    try {
-      return FunctionPrototypeCall(fn, thisArg);
-    } catch (error) {
-      return PromiseReject(error);
-    }
-  };
+  return async () => FunctionPrototypeCall(fn, thisArg);
 }
 
 // Raw variants that skip the async wrapper's implicit result promise.
@@ -382,24 +388,12 @@ function thenAlgorithmResult(result, onFulfilled, onRejected) {
 
 function createPromiseCallback1Param(name, fn, thisArg) {
   validateFunction(fn, name);
-  return (arg) => {
-    try {
-      return FunctionPrototypeCall(fn, thisArg, arg);
-    } catch (error) {
-      return PromiseReject(error);
-    }
-  };
+  return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
 }
 
 function createPromiseCallback2Params(name, fn, thisArg) {
   validateFunction(fn, name);
-  return (arg1, arg2) => {
-    try {
-      return FunctionPrototypeCall(fn, thisArg, arg1, arg2);
-    } catch (error) {
-      return PromiseReject(error);
-    }
-  };
+  return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
 }
 
 function isPromisePending(promise) {
@@ -408,31 +402,12 @@ function isPromisePending(promise) {
   return details?.[0] === kPending;
 }
 
-// Convert a promise-returning algorithm's raw result into a Promise. A
-// value that cannot be a thenable (null, undefined, or a non-object
-// non-function primitive) becomes the shared resolved promise. Objects
-// and functions go through PromiseResolve so a `.then` lookup, if any,
-// stays observable.
-function promiseFromAlgorithmResult(result) {
-  if (isNonThenable(result))
-    return kResolvedPromise;
-  return PromiseResolve(result);
-}
-
-// Cancel/flush/abort only: insert an extra microtask so "upon fulfillment"
-// of an already-settled user promise runs after start-settlement reactions
-// queued during construction. Pull/write must not use this.
-function delayedAlgorithmResult(result) {
-  if (isNonThenable(result))
-    return kResolvedPromise;
-  return PromisePrototypeThen(kResolvedPromise, () => result);
-}
-
 // Shared shapes for lazily-materialized { promise, resolve, reject }
-// records whose settlement is already known.
+// records whose settlement is already known. Each call mints a fresh
+// promise so public slots (writer.ready / writer.closed) stay distinct.
 function resolvedRecord() {
   return {
-    promise: kResolvedPromise,
+    promise: PromiseResolve(),
     resolve: undefined,
     reject: undefined,
   };
@@ -456,13 +431,16 @@ function setPromiseHandled(promise) {
   PromisePrototypeThen(promise, undefined, () => {});
 }
 
-// Shared no-op. Start/pull/write use the raw-callback contract (see
-// createRawCallback*): a non-thenable return takes the allocation-free
-// path in thenAlgorithmResult(). Cancel/flush/abort wrap the result
-// with promiseFromAlgorithmResult/delayedAlgorithmResult, so a sync
-// no-op is equivalent to the previous async empty functions.
+async function nonOpFlush() {}
+
+// Shared non-op for the start/pull/write algorithm callbacks, which all
+// follow the raw-callback contract (see createRawCallback*): the
+// non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
 function nonOpCallback() {}
 
+async function nonOpCancel() {}
+
 let transfer;
 function lazyTransfer() {
   if (transfer === undefined)
@@ -478,7 +456,6 @@ module.exports = {
   Queue,
   canCopyArrayBuffer,
   cloneAsUint8Array,
-  isNonThenable,
   copyArrayBuffer,
   createPromiseCallbackNoParams,
   createPromiseCallback1Param,
@@ -493,6 +470,7 @@ module.exports = {
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  isNonThenable,
   isPromisePending,
   kEmptyQueue,
   kResolvedPromise,
@@ -501,10 +479,8 @@ module.exports = {
   lazyTransfer,
   materializeQueue,
   nonOpCallback,
-  nonOpCancel: nonOpCallback,
-  nonOpFlush: nonOpCallback,
-  promiseFromAlgorithmResult,
-  delayedAlgorithmResult,
+  nonOpCancel,
+  nonOpFlush,
   peekQueueValue,
   rejectedHandledRecord,
   resetQueue,