From 6491a1857c183cfbda4695fc00970f22da2919e2 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 22 Oct 2026 03:09:10 +0200 Subject: [PATCH 1/2] Revert "stream: avoid promise allocation for parked transform writes" This reverts commit bbb00558c592fe4797364d08be392d645f930af4. 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: https://github.com/nodejs/node/pull/65625 Refs: https://github.com/nodejs/node/issues/66906 Signed-off-by: Matteo Collina --- lib/internal/webstreams/transformstream.js | 44 ++++++++++------------ 1 file changed, 19 insertions(+), 25 deletions(-) diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 85c118474894..7d6de60cf98c 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/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); } From beeec2edd9cd0da526df72f001bb28133d44445f Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 21 Oct 2026 23:46:41 +0200 Subject: [PATCH 2/2] Revert "stream: decouple transform backpressure changes" This reverts commit 86d1196ebf410ed43bc922e7ca3ef09ee01d6e39. 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: https://github.com/nodejs/node/pull/65143 Refs: https://github.com/nodejs/node/issues/66906 Signed-off-by: Matteo Collina --- lib/internal/webstreams/transformstream.js | 147 ++++++--------------- 1 file changed, 43 insertions(+), 104 deletions(-) diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 7d6de60cf98c..72741199f7ed 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/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) {