Skip to content

More deduplication and cleanup in TS streams #7795

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jasnell wants to merge 2 commits into from
Open

More deduplication and cleanup in TS streams #7795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6094cc6
Drop TransformStream guards that construction makes unreachable
jasnell Oct 16, 2026
7f02098
Run TransformStream's close, abort and cancel through one finish
jasnell Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/per_isolate/webstreams/readable.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4995,8 +4995,8 @@ module.exports = {
getReadableStreamGetState(stream),
getStoredError: <R>(stream: ReadableStream<R>) =>
getReadableStreamStoredError(stream),
// ReadableStreamDefaultControllerCanCloseOrEnqueue, for the
// TransformStream controller's enqueue pre-check.
// ReadableStreamDefaultControllerCanCloseOrEnqueue, which TransformStream
// checks before it enqueues into its readable and before it closes it.
canCloseOrEnqueue: <R>(controller: object) =>
defaultControllerCanCloseOrEnqueue(
controller as ReadableStreamDefaultController<R>
Expand Down
212 changes: 65 additions & 147 deletions src/per_isolate/webstreams/transform.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,11 +152,6 @@ function makeTransformerAlgorithms<I, O>(

// ---------------------------------------------------------------------------

let transformStreamDefaultControllerInit: <I, O>(
controller: TransformStreamDefaultController<I, O>,
stream: TransformStream<I, O>
) => void;

let transformStreamEnqueue: <I, O>(
stream: TransformStream<I, O>,
chunk: O
Expand All @@ -178,7 +173,7 @@ class TransformStreamDefaultController<
I = unknown,
O = unknown,
> implements TransformStreamDefaultControllerType<O> {
#stream: TransformStream<I, O> | undefined;
#stream: TransformStream<I, O>;

static {
assertIsTransformStreamDefaultController = function <I, O>(
Expand All @@ -187,50 +182,31 @@ class TransformStreamDefaultController<
if (!isActualObject(self) || !(#stream in self))
throw new TypeError('Illegal invocation');
};

transformStreamDefaultControllerInit = (controller, stream) => {
controller.#stream = stream;
};
}

constructor(privateSymbol: symbol) {
constructor(privateSymbol: symbol, stream: TransformStream<I, O>) {
assertPrivateSymbol(privateSymbol);
this.#stream = stream;
}

get desiredSize(): number | null {
assertIsTransformStreamDefaultController(this);
const stream = this.#stream;
if (stream === undefined) {
throw new TypeError('Controller is not attached to a stream');
}
return transformStreamDesiredSize(stream);
return transformStreamDesiredSize(this.#stream);
}

enqueue(chunk: O = undefined as O): void {
assertIsTransformStreamDefaultController(this);
const stream = this.#stream;
if (stream === undefined) {
throw new TypeError('Controller is not attached to a stream');
}
transformStreamEnqueue(stream, chunk);
transformStreamEnqueue(this.#stream, chunk);
}

error(reason: unknown = undefined): void {
assertIsTransformStreamDefaultController(this);
const stream = this.#stream;
if (stream === undefined) {
throw new TypeError('Controller is not attached to a stream');
}
transformStreamError(stream, reason);
transformStreamError(this.#stream, reason);
}

terminate(): void {
assertIsTransformStreamDefaultController(this);
const stream = this.#stream;
if (stream === undefined) {
throw new TypeError('Controller is not attached to a stream');
}
transformStreamTerminate(stream);
transformStreamTerminate(this.#stream);
}
}

Expand All @@ -241,8 +217,10 @@ class TransformStream<I = unknown, O = unknown> {
// allocated when there are no transformer callbacks to receive it.
// @ts-expect-error -- retained for debugging/future use
#controller?: TransformStreamDefaultController<I, O> | undefined;
#readableController: object | undefined;
#writableController: object | undefined;
// Recorded by the inner streams' start algorithms, which run while the
// constructor builds the pair, before any user code can reach it.
#readableController!: object;
#writableController!: object;
// Backpressure starts ON: the readable side must pull once before the
// sink transforms the first chunk.
#backpressure: boolean = true;
Expand All @@ -257,21 +235,14 @@ class TransformStream<I = unknown, O = unknown> {
#finishPromise: Promise<void> | undefined;

static {
transformStreamDesiredSize = (stream) => {
const readableController = stream.#readableController;
return readableController === undefined
? null
: readableControllerDesiredSizeGet(readableController);
};
transformStreamDesiredSize = (stream) =>
readableControllerDesiredSizeGet(stream.#readableController);

transformStreamEnqueue = <I, O>(
stream: TransformStream<I, O>,
chunk: O
) => {
const readableController = stream.#readableController;
if (readableController === undefined) {
throw new TypeError('TransformStream is not fully initialized');
}
// Spec TransformStreamDefaultControllerEnqueue step 4:
// if ReadableStreamDefaultControllerCanCloseOrEnqueue is false
// (close requested, or the readable no longer readable), throw a
Expand Down Expand Up @@ -308,22 +279,12 @@ class TransformStream<I = unknown, O = unknown> {
stream: TransformStream<I, O>,
reason: unknown
) => {
const readableController = stream.#readableController;
if (readableController !== undefined) {
readableControllerError(readableController, reason);
}
readableControllerError(stream.#readableController, reason);
stream.#errorWritableAndUnblockWrite(reason);
};

transformStreamTerminate = <I, O>(stream: TransformStream<I, O>) => {
const readableController = stream.#readableController;
if (readableController !== undefined) {
try {
readableControllerClose(readableController);
} catch {
// Already closed or errored — nothing to do.
}
}
stream.#closeReadable();
stream.#errorWritableAndUnblockWrite(
new TypeError('The transform stream has been terminated')
);
Expand All @@ -348,105 +309,63 @@ class TransformStream<I = unknown, O = unknown> {
this.#flushAlgorithm = undefined;
}

// Spec: TransformStreamDefaultSinkCloseAlgorithm.
#sinkClose(): Promise<void> {
if (this.#finishPromise !== undefined) return this.#finishPromise;
const { promise, resolve, reject } =
PromiseWithResolvers() as PromiseWithResolversType<void>;
this.#finishPromise = promise;
const flushAlgorithm = this.#flushAlgorithm;
const flushResult =
flushAlgorithm !== undefined ? flushAlgorithm() : PromiseResolve();
this.#clearAlgorithms();
markPromiseHandled(
PromisePrototypeThen(
flushResult,
() => {
if (readableInternals.getState(this.#readable) === 'errored') {
reject(readableInternals.getStoredError(this.#readable));
return;
}
const rc = this.#readableController;
if (rc !== undefined) {
try {
readableControllerClose(rc);
} catch {
// Already closed — acceptable per spec.
}
}
resolve();
},
(r: unknown) => {
const rc = this.#readableController;
if (rc !== undefined) readableControllerError(rc, r);
reject(r);
}
)
);
return promise;
// Spec: ReadableStreamDefaultControllerClose, which returns at once when
// the readable can no longer close or enqueue (close requested, closed or
// errored); the captured close() throws then.
#closeReadable(): void {
const readableController = this.#readableController;
if (readableInternals.canCloseOrEnqueue(readableController)) {
readableControllerClose(readableController);
}
}

// Spec: TransformStreamDefaultSinkAbortAlgorithm.
#sinkAbort(reason: unknown): Promise<void> {
// Spec: TransformStreamDefaultSinkCloseAlgorithm ('close'),
// TransformStreamDefaultSinkAbortAlgorithm ('abort') and
// TransformStreamDefaultSourceCancelAlgorithm ('cancel'). They differ in
// the algorithm they run, flush or cancel, and in the half they end: the
// sink's close and abort end the readable, the source's cancel the
// writable. Whichever runs first sets the finish promise before its
// algorithm runs, so one started from inside flush() or cancel() returns
// it, as does any started later. The ended half's state is read when the
// algorithm settles (step 7.1.1): an abort, a terminate() or an error that
// reached it by then rejects the finish with its stored error.
#finish(kind: 'close' | 'abort' | 'cancel', reason?: unknown): Promise<void> {
if (this.#finishPromise !== undefined) return this.#finishPromise;
const { promise, resolve, reject } =
PromiseWithResolvers() as PromiseWithResolversType<void>;
this.#finishPromise = promise;
const cancelAlgorithm = this.#cancelAlgorithm;
const cancelResult =
cancelAlgorithm !== undefined
? cancelAlgorithm(reason)
: PromiseResolve();
const algorithm: ((reason: unknown) => Promise<void>) | undefined =
kind === 'close' ? this.#flushAlgorithm : this.#cancelAlgorithm;
const result =
algorithm !== undefined ? algorithm(reason) : PromiseResolve();
this.#clearAlgorithms();
markPromiseHandled(
PromisePrototypeThen(
cancelResult,
() => {
if (readableInternals.getState(this.#readable) === 'errored') {
reject(readableInternals.getStoredError(this.#readable));
return;
}
const rc = this.#readableController;
if (rc !== undefined) readableControllerError(rc, reason);
resolve();
},
(r: unknown) => {
const rc = this.#readableController;
if (rc !== undefined) readableControllerError(rc, r);
reject(r);
const endsWritable = kind === 'cancel';
const internals = endsWritable ? writableInternals : readableInternals;
const ended = endsWritable ? this.#writable : this.#readable;
const errorEnded = endsWritable
? (e: unknown): void => {
this.#errorWritableAndUnblockWrite(e);
}
)
);
return promise;
}

// Spec: TransformStreamDefaultSourceCancelAlgorithm. The writable's state
// is read when the cancel algorithm settles (step 7.1.1): an abort, a
// terminate() or an error that reached it by then rejects the cancel.
#sourceCancel(reason: unknown): Promise<void> {
if (this.#finishPromise !== undefined) return this.#finishPromise;
const { promise, resolve, reject } =
PromiseWithResolvers() as PromiseWithResolversType<void>;
this.#finishPromise = promise;
const cancelAlgorithm = this.#cancelAlgorithm;
const cancelResult =
cancelAlgorithm !== undefined
? cancelAlgorithm(reason)
: PromiseResolve();
this.#clearAlgorithms();
: (e: unknown): void => {
readableControllerError(this.#readableController, e);
};
markPromiseHandled(
PromisePrototypeThen(
cancelResult,
result,
() => {
if (writableInternals.getState(this.#writable) === 'errored') {
reject(writableInternals.getStoredError(this.#writable));
if (internals.getState(ended) === 'errored') {
reject(internals.getStoredError(ended));
return;
}
this.#errorWritableAndUnblockWrite(reason);
if (kind === 'close') {
this.#closeReadable();
} else {
errorEnded(reason);
}
resolve();
},
(r: unknown) => {
this.#errorWritableAndUnblockWrite(r);
errorEnded(r);
reject(r);
}
)
Expand All @@ -460,10 +379,7 @@ class TransformStream<I = unknown, O = unknown> {
// later readable.cancel() from invoking the transformer's cancel
// callback after the stream has already errored/terminated.
this.#clearAlgorithms();
const writableController = this.#writableController;
if (writableController !== undefined) {
writableControllerError(writableController, reason);
}
writableControllerError(this.#writableController, reason);
if (this.#backpressure) {
this.#setBackpressure(false);
}
Expand Down Expand Up @@ -558,8 +474,10 @@ class TransformStream<I = unknown, O = unknown> {
} else {
// ---- STANDARD PATH (transformer has algorithms) ----

controller = new TransformStreamDefaultController<I, O>(kPrivateSymbol);
transformStreamDefaultControllerInit(controller, this);
controller = new TransformStreamDefaultController<I, O>(
kPrivateSymbol,
this
);
this.#controller = controller;

const algorithms = makeTransformerAlgorithms(
Expand Down Expand Up @@ -632,8 +550,8 @@ class TransformStream<I = unknown, O = unknown> {
return startPromise;
},
write: sinkWrite,
close: () => this.#sinkClose(),
abort: (reason: unknown) => this.#sinkAbort(reason),
close: () => this.#finish('close'),
abort: (reason: unknown) => this.#finish('abort', reason),
},
// Converted already: the inner stream reads its own data properties,
// and its default highWaterMark (1) is the TransformStream's.
Expand All @@ -652,7 +570,7 @@ class TransformStream<I = unknown, O = unknown> {
return startPromise;
},
pull: sourcePull,
cancel: (reason: unknown) => this.#sourceCancel(reason),
cancel: (reason: unknown) => this.#finish('cancel', reason),
},
{
__proto__: null,
Expand All @@ -667,7 +585,7 @@ class TransformStream<I = unknown, O = unknown> {
// for bodies built from this transform; not enforced here.
if (expectedLength !== undefined) {
readableInternals.setControllerExpectedLength(
this.#readableController as object,
this.#readableController,
expectedLength
);
}
Expand Down
Loading