Revert "stream: decouple transform backpressure changes"

This reverts commit 86d1196.

Resolving a parked write from the readable side ran the first parked
transform() in the reader's async context. Conflicts with later
commits are resolved by keeping their state record classes and the
kParkedAlgorithmResult check used by ReadableStream.from().

Refs: #65143
Refs: #66906

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/webstreams/transformstream.js

@@ -5,8 +5,6 @@ const {
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
   PromisePrototypeThen,
-  PromiseReject,
-  PromiseResolve,
   PromiseWithResolvers,
   Symbol,
   SymbolToStringTag,
@@ -46,14 +44,12 @@ const {
 
 const {
   createPromiseCallback1Param,
-  createRawCallback2Params,
+  createPromiseCallback2Params,
   customInspect,
   extractHighWaterMark,
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
-  kParkedAlgorithmResult,
-  kResolvedPromise,
   kState,
   kType,
   nonOpCancel,
@@ -261,12 +257,7 @@ class TransformStreamState {
   writable = undefined;
   controller = undefined;
   backpressure = undefined;
-  // Continuation slots replacing the spec's
-  // [[backpressureChangePromise]]; see transformStreamSetBackpressure.
-  pullPending = false;
-  pendingWrite = undefined;
-  pendingWriteChunk = undefined;
-  writeContinuation = undefined;
+  backpressureChange = undefined;
 }
 ObjectSetPrototypeOf(TransformStreamState.prototype, null);
 
@@ -372,9 +363,7 @@ const isTransformStream =
 const isTransformStreamDefaultController =
   isBrandCheck('TransformStreamDefaultController');
 
-// Raw callback (see createRawCallback*): invoked inside the try/catch of
-// transformStreamDefaultControllerPerformTransform.
-function defaultTransformAlgorithm(chunk, controller) {
+async function defaultTransformAlgorithm(chunk, controller) {
   transformStreamDefaultControllerEnqueue(controller, chunk);
 }
 
@@ -444,30 +433,24 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Both observers are internal, so the
-// promise record is replaced by continuation slots: a parked pull is
-// completed by delivering the readable controller's pull-fulfilled step,
-// and a parked write by the cached write continuation (see
-// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
-// shared resolved promise at the exact microtask position the old
-// record's reaction would have had.
+// backpressure next becomes false). Instead of allocating a fresh promise
+// record on every flip, the record is materialized lazily on first
+// observation and dropped once settled; flips nobody is waiting on
+// allocate nothing.
+function transformStreamBackpressureChangePromise(stream) {
+  const state = stream[kState];
+  return (state.backpressureChange ??= PromiseWithResolvers()).promise;
+}
+
 function transformStreamSetBackpressure(stream, backpressure) {
   const state = stream[kState];
   assert(state.backpressure !== backpressure);
-  state.backpressure = backpressure;
-  if (backpressure) {
-    if (state.pullPending) {
-      state.pullPending = false;
-      // The pull-fulfilled step exists: a pull parked it (see
-      // transformStreamDefaultSourcePullAlgorithm), and the readable
-      // controller creates it before invoking the pull algorithm.
-      PromisePrototypeThen(
-        kResolvedPromise,
-        state.readable[kState].controller[kState].pullFulfilled);
-    }
-  } else if (state.pendingWrite !== undefined) {
-    PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
+  const backpressureChange = state.backpressureChange;
+  if (backpressureChange !== undefined) {
+    state.backpressureChange = undefined;
+    backpressureChange.resolve();
   }
+  state.backpressure = backpressure;
 }
 
 function setupTransformStreamDefaultController(
@@ -495,7 +478,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const flush = transformer?.flush;
   const cancel = transformer?.cancel;
   const transformAlgorithm = transform ?
-    createRawCallback2Params('transformer.transform', transform, transformer) :
+    createPromiseCallback2Params('transformer.transform', transform, transformer) :
     defaultTransformAlgorithm;
   const flushAlgorithm = flush ?
     createPromiseCallback1Param('transformer.flush', flush, transformer) :
@@ -548,40 +531,18 @@ function transformStreamDefaultControllerError(controller, error) {
   transformStreamError(controller[kState].stream, error);
 }
 
-// Mirrors the reference implementation's
-// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
-// the returned promise settles one microtask after the (coerced) result
-// does, and a rejection errors the transform stream before propagating.
-// The raw transform callback plus the shared resolved promise for
-// non-thenable results replace the previous async wrapper's two implicit
-// promises per chunk.
-function transformStreamDefaultControllerPerformTransform(controller, chunk) {
-  const controllerState = controller[kState];
-  const transformAlgorithm = controllerState.transformAlgorithm;
-  if (transformAlgorithm === undefined) {
-    // Algorithms were cleared by a concurrent cancel/abort/close.
-    return kResolvedPromise;
-  }
-  let result;
+async function transformStreamDefaultControllerPerformTransform(controller, chunk) {
   try {
-    result = transformAlgorithm(chunk, controller);
+    const transformAlgorithm = controller[kState].transformAlgorithm;
+    if (transformAlgorithm === undefined) {
+      // Algorithms were cleared by a concurrent cancel/abort/close.
+      return;
+    }
+    return await transformAlgorithm(chunk, controller);
   } catch (error) {
-    result = PromiseReject(error);
-  }
-  if (result === null ||
-      (typeof result !== 'object' && typeof result !== 'function')) {
-    result = kResolvedPromise;
-  } else {
-    result = PromiseResolve(result);
-  }
-  controllerState.performTransformRejected ??= (error) => {
     transformStreamError(controller[kState].stream, error);
     throw error;
-  };
-  return PromisePrototypeThen(
-    result,
-    undefined,
-    controllerState.performTransformRejected);
+  }
 }
 
 function transformStreamDefaultControllerTerminate(controller) {
@@ -602,42 +563,26 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
-  const state = stream[kState];
   const {
     writable,
     controller,
-  } = state;
+  } = stream[kState];
   assert(writable[kState].state === 'writable');
-  if (state.backpressure) {
-    // Park the chunk and one promise record; the backpressure -> false
-    // flip delivers the cached continuation (see
-    // transformStreamSetBackpressure) at the same microtask position as
-    // the old [[backpressureChangePromise]] reaction. The continuation
-    // resolves the sink promise with the perform-transform promise, so
-    // adoption reproduces the old derived-chain settle depth exactly.
-    // The writable dispatches a single write at a time, so one pending
-    // slot suffices.
-    assert(state.pendingWrite === undefined);
-    const pendingWrite = PromiseWithResolvers();
-    state.pendingWrite = pendingWrite;
-    state.pendingWriteChunk = chunk;
-    state.writeContinuation ??= () => {
-      const pending = state.pendingWrite;
-      const pendingChunk = state.pendingWriteChunk;
-      state.pendingWrite = undefined;
-      state.pendingWriteChunk = undefined;
-      const writableState = state.writable[kState];
-      if (writableState.state === 'erroring') {
-        pending.reject(writableState.storedError);
-        return;
-      }
-      assert(writableState.state === 'writable');
-      pending.resolve(
-        transformStreamDefaultControllerPerformTransform(
+  if (stream[kState].backpressure) {
+    const backpressureChange = transformStreamBackpressureChangePromise(stream);
+    return PromisePrototypeThen(
+      backpressureChange,
+      () => {
+        const {
+          writable,
+        } = stream[kState];
+        if (writable[kState].state === 'erroring')
+          throw writable[kState].storedError;
+        assert(writable[kState].state === 'writable');
+        return transformStreamDefaultControllerPerformTransform(
           controller,
-          pendingChunk));
-    };
-    return pendingWrite.promise;
+          chunk);
+      });
   }
   return transformStreamDefaultControllerPerformTransform(controller, chunk);
 }
@@ -707,15 +652,9 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 function transformStreamDefaultSourcePullAlgorithm(stream) {
-  const state = stream[kState];
-  assert(state.backpressure);
+  assert(stream[kState].backpressure);
   transformStreamSetBackpressure(stream, false);
-  // Park the pull: the next backpressure -> true flip delivers the
-  // pull-fulfilled step (see transformStreamSetBackpressure). The old
-  // [[backpressureChangePromise]] this replaces was only ever resolved,
-  // so the parked pull needs no rejection delivery.
-  state.pullPending = true;
-  return kParkedAlgorithmResult;
+  return transformStreamBackpressureChangePromise(stream);
 }
 
 function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {