Skip to content

Commit ac998a5

Browse files
mcollinaclaude
andcommitted
Revert "stream: decouple transform backpressure changes"
This reverts commit 9ec9383. 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: #65143 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 dc9ccff commit ac998a5

1 file changed

Lines changed: 43 additions & 103 deletions

File tree

‎lib/internal/webstreams/transformstream.js‎

Lines changed: 43 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@ const {
55
ObjectDefineProperties,
66
ObjectSetPrototypeOf,
77
PromisePrototypeThen,
8-
PromiseReject,
9-
PromiseResolve,
108
PromiseWithResolvers,
119
Symbol,
1210
SymbolToStringTag,
@@ -46,13 +44,12 @@ const {
4644

4745
const {
4846
createPromiseCallback1Param,
49-
createRawCallback2Params,
47+
createPromiseCallback2Params,
5048
customInspect,
5149
extractHighWaterMark,
5250
extractSizeAlgorithm,
5351
getNonWritablePropertyDescriptor,
5452
isBrandCheck,
55-
kParkedAlgorithmResult,
5653
kResolvedPromise,
5754
kState,
5855
kType,
@@ -262,12 +259,7 @@ class TransformStreamState {
262259
writable = undefined;
263260
controller = undefined;
264261
backpressure = undefined;
265-
// Continuation slots replacing the spec's
266-
// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
267-
pullPending = false;
268-
pendingWrite = undefined;
269-
pendingWriteChunk = undefined;
270-
writeContinuation = undefined;
262+
backpressureChange = undefined;
271263
}
272264
ObjectSetPrototypeOf(TransformStreamState.prototype, null);
273265

@@ -374,9 +366,7 @@ const isTransformStream =
374366
const isTransformStreamDefaultController =
375367
isBrandCheck('TransformStreamDefaultController');
376368

377-
// Raw callback (see createRawCallback*): invoked inside the try/catch of
378-
// transformStreamDefaultControllerPerformTransform.
379-
function defaultTransformAlgorithm(chunk, controller) {
369+
async function defaultTransformAlgorithm(chunk, controller) {
380370
transformStreamDefaultControllerEnqueue(controller, chunk);
381371
}
382372

@@ -452,30 +442,24 @@ function transformStreamUnblockWrite(stream) {
452442
// The spec's [[backpressureChangePromise]] is only ever observed by the
453443
// source pull algorithm (settles when backpressure next becomes true) and
454444
// by a sink write arriving while backpressure is set (settles when
455-
// backpressure next becomes false). Both observers are internal, so the
456-
// promise record is replaced by continuation slots: a parked pull is
457-
// completed by delivering the readable controller's pull-fulfilled step,
458-
// and a parked write by the cached write continuation (see
459-
// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
460-
// shared resolved promise at the exact microtask position the old
461-
// record's reaction would have had.
445+
// backpressure next becomes false). Instead of allocating a fresh promise
446+
// record on every flip, the record is materialized lazily on first
447+
// observation and dropped once settled; flips nobody is waiting on
448+
// allocate nothing.
449+
function transformStreamBackpressureChangePromise(stream) {
450+
const state = stream[kState];
451+
return (state.backpressureChange ??= PromiseWithResolvers()).promise;
452+
}
453+
462454
function transformStreamSetBackpressure(stream, backpressure) {
463455
const state = stream[kState];
464456
assert(state.backpressure !== backpressure);
465-
state.backpressure = backpressure;
466-
if (backpressure) {
467-
if (state.pullPending) {
468-
state.pullPending = false;
469-
// The pull-fulfilled step exists: a pull parked it (see
470-
// transformStreamDefaultSourcePullAlgorithm), and the readable
471-
// controller creates it before invoking the pull algorithm.
472-
PromisePrototypeThen(
473-
kResolvedPromise,
474-
state.readable[kState].controller[kState].pullFulfilled);
475-
}
476-
} else if (state.pendingWrite !== undefined) {
477-
PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
457+
const backpressureChange = state.backpressureChange;
458+
if (backpressureChange !== undefined) {
459+
state.backpressureChange = undefined;
460+
backpressureChange.resolve();
478461
}
462+
state.backpressure = backpressure;
479463
}
480464

481465
function setupTransformStreamDefaultController(
@@ -503,7 +487,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
503487
const flush = transformer?.flush;
504488
const cancel = transformer?.cancel;
505489
const transformAlgorithm = transform ?
506-
createRawCallback2Params('transformer.transform', transform, transformer) :
490+
createPromiseCallback2Params('transformer.transform', transform, transformer) :
507491
defaultTransformAlgorithm;
508492
const flushAlgorithm = flush ?
509493
createPromiseCallback1Param('transformer.flush', flush, transformer) :
@@ -556,40 +540,18 @@ function transformStreamDefaultControllerError(controller, error) {
556540
transformStreamError(controller[kState].stream, error);
557541
}
558542

559-
// Mirrors the reference implementation's
560-
// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
561-
// the returned promise settles one microtask after the (coerced) result
562-
// does, and a rejection errors the transform stream before propagating.
563-
// The raw transform callback plus the shared resolved promise for
564-
// non-thenable results replace the previous async wrapper's two implicit
565-
// promises per chunk.
566-
function transformStreamDefaultControllerPerformTransform(controller, chunk) {
567-
const controllerState = controller[kState];
568-
const transformAlgorithm = controllerState.transformAlgorithm;
569-
if (transformAlgorithm === undefined) {
570-
// Algorithms were cleared by a concurrent cancel/abort/close.
571-
return kResolvedPromise;
572-
}
573-
let result;
543+
async function transformStreamDefaultControllerPerformTransform(controller, chunk) {
574544
try {
575-
result = transformAlgorithm(chunk, controller);
545+
const transformAlgorithm = controller[kState].transformAlgorithm;
546+
if (transformAlgorithm === undefined) {
547+
// Algorithms were cleared by a concurrent cancel/abort/close.
548+
return;
549+
}
550+
return await transformAlgorithm(chunk, controller);
576551
} catch (error) {
577-
result = PromiseReject(error);
578-
}
579-
if (result === null ||
580-
(typeof result !== 'object' && typeof result !== 'function')) {
581-
result = kResolvedPromise;
582-
} else {
583-
result = PromiseResolve(result);
584-
}
585-
controllerState.performTransformRejected ??= (error) => {
586552
transformStreamError(controller[kState].stream, error);
587553
throw error;
588-
};
589-
return PromisePrototypeThen(
590-
result,
591-
undefined,
592-
controllerState.performTransformRejected);
554+
}
593555
}
594556

595557
function transformStreamDefaultControllerTerminate(controller) {
@@ -610,42 +572,26 @@ function transformStreamDefaultControllerTerminate(controller) {
610572
}
611573

612574
function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
613-
const state = stream[kState];
614575
const {
615576
writable,
616577
controller,
617-
} = state;
578+
} = stream[kState];
618579
assert(writable[kState].state === 'writable');
619-
if (state.backpressure) {
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;
631-
state.pendingWriteChunk = chunk;
632-
state.writeContinuation ??= () => {
633-
const pending = state.pendingWrite;
634-
const pendingChunk = state.pendingWriteChunk;
635-
state.pendingWrite = undefined;
636-
state.pendingWriteChunk = undefined;
637-
const writableState = state.writable[kState];
638-
if (writableState.state === 'erroring') {
639-
pending.reject(writableState.storedError);
640-
return;
641-
}
642-
assert(writableState.state === 'writable');
643-
pending.resolve(
644-
transformStreamDefaultControllerPerformTransform(
580+
if (stream[kState].backpressure) {
581+
const backpressureChange = transformStreamBackpressureChangePromise(stream);
582+
return PromisePrototypeThen(
583+
backpressureChange,
584+
() => {
585+
const {
586+
writable,
587+
} = stream[kState];
588+
if (writable[kState].state === 'erroring')
589+
throw writable[kState].storedError;
590+
assert(writable[kState].state === 'writable');
591+
return transformStreamDefaultControllerPerformTransform(
645592
controller,
646-
pendingChunk));
647-
};
648-
return pendingWrite.promise;
593+
chunk);
594+
});
649595
}
650596
return transformStreamDefaultControllerPerformTransform(controller, chunk);
651597
}
@@ -715,15 +661,9 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
715661
}
716662

717663
function transformStreamDefaultSourcePullAlgorithm(stream) {
718-
const state = stream[kState];
719-
assert(state.backpressure);
664+
assert(stream[kState].backpressure);
720665
transformStreamSetBackpressure(stream, false);
721-
// Park the pull: the next backpressure -> true flip delivers the
722-
// pull-fulfilled step (see transformStreamSetBackpressure). The old
723-
// [[backpressureChangePromise]] this replaces was only ever resolved,
724-
// so the parked pull needs no rejection delivery.
725-
state.pullPending = true;
726-
return kParkedAlgorithmResult;
666+
return transformStreamBackpressureChangePromise(stream);
727667
}
728668

729669
function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {

0 commit comments

Comments
 (0)