Revert "stream: avoid promise allocation for parked transform writes"

This reverts commit bbb0055.

A parked TransformStream write was resumed from the read that lifted
the backpressure, so transform(), the writes queued behind it and
flush() ran in the reader's async context instead of the writer's,
losing AsyncLocalStorage state.

The parked write error test is kept, as it covers spec behavior that
still holds after the revert.

Refs: #65625
Refs: #66906

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

Author: mcollina

lib/internal/webstreams/transformstream.js

@@ -264,7 +264,7 @@ class TransformStreamState {
   // Continuation slots replacing the spec's
   // [[backpressureChangePromise]]; see transformStreamSetBackpressure.
   pullPending = false;
-  pendingWriteParked = false;
+  pendingWrite = undefined;
   pendingWriteChunk = undefined;
   writeContinuation = undefined;
 }
@@ -465,7 +465,7 @@ function transformStreamSetBackpressure(stream, backpressure) {
         kResolvedPromise,
         state.readable[kState].controller[kState].pullFulfilled);
     }
-  } else if (state.pendingWriteParked) {
+  } else if (state.pendingWrite !== undefined) {
     PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
   }
 }
@@ -609,41 +609,35 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
   } = state;
   assert(writable[kState].state === 'writable');
   if (state.backpressure) {
-    // Park the chunk; the backpressure -> false flip delivers the cached
-    // continuation (see transformStreamSetBackpressure) at the same
-    // microtask position as the old [[backpressureChangePromise]]
-    // reaction. The continuation completes the parked write by wiring
-    // the perform-transform promise directly to the writable
-    // controller's write reactions (they exist: the controller creates
-    // them before invoking the write algorithm), replacing the promise
-    // record the old code allocated and resolved per parked chunk. The
-    // writable dispatches a single write at a time, so one pending slot
-    // suffices.
-    assert(!state.pendingWriteParked);
-    state.pendingWriteParked = true;
+    // 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.pendingWriteParked = false;
+      state.pendingWrite = undefined;
       state.pendingWriteChunk = undefined;
       const writableState = state.writable[kState];
-      const writableControllerState = writableState.controller[kState];
       if (writableState.state === 'erroring') {
-        const error = writableState.storedError;
-        PromisePrototypeThen(
-          kResolvedPromise,
-          () => writableControllerState.writeRejected(error));
+        pending.reject(writableState.storedError);
         return;
       }
       assert(writableState.state === 'writable');
-      PromisePrototypeThen(
+      pending.resolve(
         transformStreamDefaultControllerPerformTransform(
           controller,
-          pendingChunk),
-        writableControllerState.writeFulfilled,
-        writableControllerState.writeRejected);
+          pendingChunk));
     };
-    return kParkedAlgorithmResult;
+    return pendingWrite.promise;
   }
   return transformStreamDefaultControllerPerformTransform(controller, chunk);
 }