Skip to content

Commit dc9ccff

Browse files
mcollinaclaude
andcommitted
Revert "stream: avoid promise allocation for parked transform writes"
This reverts commit f81a148. 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 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LGUJ9PwPYkYJiCpUEsr7KJ Signed-off-by: Matteo Collina <hello@matteocollina.com>
1 parent 6466cb1 commit dc9ccff

1 file changed

Lines changed: 19 additions & 25 deletions

File tree

‎lib/internal/webstreams/transformstream.js‎

Lines changed: 19 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -265,7 +265,7 @@ class TransformStreamState {
265265
// Continuation slots replacing the spec's
266266
// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
267267
pullPending = false;
268-
pendingWriteParked = false;
268+
pendingWrite = undefined;
269269
pendingWriteChunk = undefined;
270270
writeContinuation = undefined;
271271
}
@@ -473,7 +473,7 @@ function transformStreamSetBackpressure(stream, backpressure) {
473473
kResolvedPromise,
474474
state.readable[kState].controller[kState].pullFulfilled);
475475
}
476-
} else if (state.pendingWriteParked) {
476+
} else if (state.pendingWrite !== undefined) {
477477
PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
478478
}
479479
}
@@ -617,41 +617,35 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
617617
} = state;
618618
assert(writable[kState].state === 'writable');
619619
if (state.backpressure) {
620-
// Park the chunk; the backpressure -> false flip delivers the cached
621-
// continuation (see transformStreamSetBackpressure) at the same
622-
// microtask position as the old [[backpressureChangePromise]]
623-
// reaction. The continuation completes the parked write by wiring
624-
// the perform-transform promise directly to the writable
625-
// controller's write reactions (they exist: the controller creates
626-
// them before invoking the write algorithm), replacing the promise
627-
// record the old code allocated and resolved per parked chunk. The
628-
// writable dispatches a single write at a time, so one pending slot
629-
// suffices.
630-
assert(!state.pendingWriteParked);
631-
state.pendingWriteParked = true;
620+
// Park the chunk and one promise record; the backpressure -> false
621+
// flip delivers the cached continuation (see
622+
// transformStreamSetBackpressure) at the same microtask position as
623+
// the old [[backpressureChangePromise]] reaction. The continuation
624+
// resolves the sink promise with the perform-transform promise, so
625+
// adoption reproduces the old derived-chain settle depth exactly.
626+
// The writable dispatches a single write at a time, so one pending
627+
// slot suffices.
628+
assert(state.pendingWrite === undefined);
629+
const pendingWrite = PromiseWithResolvers();
630+
state.pendingWrite = pendingWrite;
632631
state.pendingWriteChunk = chunk;
633632
state.writeContinuation ??= () => {
633+
const pending = state.pendingWrite;
634634
const pendingChunk = state.pendingWriteChunk;
635-
state.pendingWriteParked = false;
635+
state.pendingWrite = undefined;
636636
state.pendingWriteChunk = undefined;
637637
const writableState = state.writable[kState];
638-
const writableControllerState = writableState.controller[kState];
639638
if (writableState.state === 'erroring') {
640-
const error = writableState.storedError;
641-
PromisePrototypeThen(
642-
kResolvedPromise,
643-
() => writableControllerState.writeRejected(error));
639+
pending.reject(writableState.storedError);
644640
return;
645641
}
646642
assert(writableState.state === 'writable');
647-
PromisePrototypeThen(
643+
pending.resolve(
648644
transformStreamDefaultControllerPerformTransform(
649645
controller,
650-
pendingChunk),
651-
writableControllerState.writeFulfilled,
652-
writableControllerState.writeRejected);
646+
pendingChunk));
653647
};
654-
return kParkedAlgorithmResult;
648+
return pendingWrite.promise;
655649
}
656650
return transformStreamDefaultControllerPerformTransform(controller, chunk);
657651
}

0 commit comments

Comments
 (0)