From 6a04945f8d97843446d70d5d27c7bd9a7ca0da37 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 22 Oct 2026 03:10:10 +0200 Subject: [PATCH 1/2] Revert "stream: avoid promise allocation for parked transform writes" This reverts commit f81a1483b34e79b0fd3de5ac18e49d8a4f40ce4c. 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 81f60b284f53..05a821657970 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -265,7 +265,7 @@ class TransformStreamState { // Continuation slots replacing the spec's // [[backpressureChangePromise]]; see transformStreamSetBackpressure. pullPending = false; - pendingWriteParked = false; + pendingWrite = undefined; pendingWriteChunk = undefined; writeContinuation = undefined; } @@ -473,7 +473,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); } } @@ -617,41 +617,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 2463b52685f82ddf48a24f7de1313eb98c4f1656 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 21 Oct 2026 23:46:27 +0200 Subject: [PATCH 2/2] Revert "stream: decouple transform backpressure changes" This reverts commit 9ec9383d1d662694f8d8ffa5e15d946cc0c5514e. 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 | 146 ++++++--------------- 1 file changed, 43 insertions(+), 103 deletions(-) diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 05a821657970..d07010dd2f15 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,13 +44,12 @@ const { const { createPromiseCallback1Param, - createRawCallback2Params, + createPromiseCallback2Params, customInspect, extractHighWaterMark, extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, - kParkedAlgorithmResult, kResolvedPromise, kState, kType, @@ -262,12 +259,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); @@ -374,9 +366,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); } @@ -452,30 +442,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( @@ -503,7 +487,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) : @@ -556,40 +540,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) { @@ -610,42 +572,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); } @@ -715,15 +661,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) {