stream: avoid promise allocation for parked transform writes

A transform sink write arriving under backpressure parked the chunk
together with a PromiseWithResolvers record whose promise was returned
to the writable controller and later resolved with the perform-transform
promise. The writable's write reactions already exist before the write
algorithm runs, so the parked write can instead return the parked-result
sentinel and have the continuation wire the perform-transform promise
directly to those reactions, dropping the per-chunk promise record and
the thenable adoption hop. Failures while erroring are delivered in a
microtask, preserving the old rejection position.

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;
-  pendingWrite = undefined;
+  pendingWriteParked = false;
   pendingWriteChunk = undefined;
   writeContinuation = undefined;
 }
@@ -465,7 +465,7 @@ function transformStreamSetBackpressure(stream, backpressure) {
         kResolvedPromise,
         state.readable[kState].controller[kState].pullFulfilled);
     }
-  } else if (state.pendingWrite !== undefined) {
+  } else if (state.pendingWriteParked) {
     PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
   }
 }
@@ -609,35 +609,41 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
   } = state;
   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;
+    // 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;
     state.pendingWriteChunk = chunk;
     state.writeContinuation ??= () => {
-      const pending = state.pendingWrite;
       const pendingChunk = state.pendingWriteChunk;
-      state.pendingWrite = undefined;
+      state.pendingWriteParked = false;
       state.pendingWriteChunk = undefined;
       const writableState = state.writable[kState];
+      const writableControllerState = writableState.controller[kState];
       if (writableState.state === 'erroring') {
-        pending.reject(writableState.storedError);
+        const error = writableState.storedError;
+        PromisePrototypeThen(
+          kResolvedPromise,
+          () => writableControllerState.writeRejected(error));
         return;
       }
       assert(writableState.state === 'writable');
-      pending.resolve(
+      PromisePrototypeThen(
         transformStreamDefaultControllerPerformTransform(
           controller,
-          pendingChunk));
+          pendingChunk),
+        writableControllerState.writeFulfilled,
+        writableControllerState.writeRejected);
     };
-    return pendingWrite.promise;
+    return kParkedAlgorithmResult;
   }
   return transformStreamDefaultControllerPerformTransform(controller, chunk);
 }

test/parallel/test-whatwg-transformstream-parked-write-error.js

@@ -0,0 +1,57 @@
+'use strict';
+
+require('../common');
+const { test } = require('node:test');
+const assert = require('node:assert');
+const { TransformStream } = require('stream/web');
+const { setImmediate } = require('timers/promises');
+
+// A transform sink write arriving while the stream has backpressure is
+// parked until backpressure clears. These cases complete a parked write
+// while the writable side is already erroring.
+//
+// The setImmediate() lets the start algorithm settle so the write below
+// reaches the sink and parks (backpressure is set until the readable
+// side pulls).
+
+test('readable.cancel() rejects a parked write with the cancel reason', async () => {
+  const stream = new TransformStream();
+  const writer = stream.writable.getWriter();
+  await setImmediate();
+
+  const reason = new Error('cancelled');
+  const write = writer.write('parked');
+  await stream.readable.cancel(reason);
+  await assert.rejects(write, (err) => err === reason);
+});
+
+test('controller.error() rejects a parked write with the stored error', async () => {
+  let controller;
+  const stream = new TransformStream({
+    start(c) { controller = c; },
+  });
+  const writer = stream.writable.getWriter();
+  await setImmediate();
+
+  const reason = new Error('boom');
+  const write = writer.write('parked');
+  controller.error(reason);
+  await assert.rejects(write, (err) => err === reason);
+  await assert.rejects(writer.closed, (err) => err === reason);
+});
+
+test('controller.terminate() rejects a parked write', async () => {
+  let controller;
+  const stream = new TransformStream({
+    start(c) { controller = c; },
+  });
+  const writer = stream.writable.getWriter();
+  await setImmediate();
+
+  const write = writer.write('parked');
+  controller.terminate();
+  await assert.rejects(write, {
+    name: 'TypeError',
+    message: /terminated/,
+  });
+});