From 22357e8bedc5b5d2ab27c7a9573457366cec064a Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Wed, 28 Oct 2026 22:18:11 -0500 Subject: [PATCH 1/2] Lower ReadableStream tee, pipeTo and pipeThrough in TypeScript In TypeScript, tee(), pipeTo() and pipeThrough() on a ReadableStream were refused with SC2020, and the WritableStream constructor and a TransformStream with a transformer had no lowering, while JavaScript programs reach all of them through the web stream classes (scr_webstreams.c). A TypeScript program that uses one of them now builds its ReadableStream, WritableStream and TransformStream values with those classes, as JavaScript does, and %main installs them, so a native stream such as a Response body answers the three methods through its web view. A program that uses only getReader(), cancel(), locked and the identity transform keeps the native streams and builds to the same binary. The fallback declarations gain UnderlyingSink and Transformer, the WritableStream constructor takes a sink and the TransformStream constructor a transformer and queuing strategies, and their controllers map to checked-dynamic handles. The fetch profile lists the three methods as static with the new static-web-streams fetch fixture as their differential evidence, and keeps them in the set that links the classes. The surface manifest is regenerated, and the surface-manifest probe that kept tee() and pipeTo() fenced now fences values() and the async iterator. Adds the static-web-streams fixture and the corpus program web-streams-typed-pipes.ts, with its corpus cost and order baseline. --- .../builtin-apis/streams-and-readline.mdx | 6 +- .../ambient/scriptc-node-fallback.d.ts | 31 +++- packages/compiler/src/compat/fetch-profile.ts | 51 +++--- .../lowering/builtins/web-streams-dyn.ts | 85 +++++++++- .../src/frontend/lowering/lower-stdlib.ts | 7 +- .../compiler/src/frontend/lowering/lowerer.ts | 4 + packages/compiler/src/frontend/type-mapper.ts | 2 + packages/compiler/surface-manifest.json | 15 +- .../corpus/web-streams-typed-pipes.ts.json | 6 + tests/corpus/web-streams-typed-pipes.ts | 158 ++++++++++++++++++ .../fetch/static-web-streams/main.mts | 96 +++++++++++ .../__snapshots__/node-types-fenced.txt | 26 +-- .../web-streams-typed-pipes.ts.json | 1 + tests/harness/fetch.test.ts | 1 + tests/harness/surface-manifest.test.ts | 2 +- 15 files changed, 419 insertions(+), 72 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-pipes.ts.json create mode 100644 tests/corpus/web-streams-typed-pipes.ts create mode 100644 tests/fixtures/fetch/static-web-streams/main.mts create mode 100644 tests/harness/corpus-costs/web-streams-typed-pipes.ts.json diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index 52bf534e8..01cd9eb9d 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -35,7 +35,9 @@ `stream.compose` accepts streams and async generator functions in any position, an async function as the last stage, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it. A web `ReadableStream` is read as an async iterable by `Readable.from` and as the first stage of `compose` or `pipeline`, and through its reader by `Readable.fromWeb` (with `objectMode`, `highWaterMark`, `encoding`, and `signal` in an options object literal) and `Duplex.from`. In TypeScript, `new TransformStream()` without arguments builds the identity transform: its readable side is a web `ReadableStream`, its writable side supports `getWriter`, `abort`, `close`, and `locked`, and the writer supports `write`, `close`, `abort`, `releaseLock`, `ready`, `closed`, and `desiredSize`, with Node's backpressure (a write completes when the readable side reads its chunk). - A web stream in a later compose stage and `Readable.toWeb` are unsupported, and in TypeScript, so are a `TransformStream` with a transformer or queuing strategies, the `WritableStream` constructor, `pipeTo`, and `pipeThrough`. + In TypeScript, `tee`, `pipeTo`, and `pipeThrough`, the `WritableStream` constructor with an underlying sink, and a `TransformStream` with a transformer or queuing strategies compile: a program that uses one of them builds its `ReadableStream`, `WritableStream`, and `TransformStream` values with the same classes as JavaScript, and a `Response` or `Request` body has those members too. + A TypeScript program that uses only `getReader`, `cancel`, `locked`, and the identity `TransformStream` keeps the smaller native streams. + A web stream in a later compose stage and `Readable.toWeb` are unsupported. The `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream. `readable.compose(stage, { signal })` is `stream.compose(readable, stage)` with the signal added; the options are an object literal whose `signal` is a variable or a property read. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, with or without function stages, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. @@ -60,7 +62,7 @@ The `readline/promises` `Readline` class writes to the same outputs. `readline.emitKeypressEvents` emits `'keypress'` events on `process.stdin` or a readable stream, shared with terminal interfaces on that stream as in Node; a listener's first parameter must admit `undefined`, which it receives for escape sequences. Chained listener registration is unsupported. - `node:stream/web` exports the same classes as the globals; in TypeScript, only `ReadableStream` and the identity `TransformStream` have lowerings. + `node:stream/web` exports the same classes as the globals; in TypeScript, `ReadableStream`, `WritableStream`, and `TransformStream` have lowerings. In JavaScript, `ReadableStream`, `WritableStream`, and `TransformStream` with their default readers, writers, and controllers, `CountQueuingStrategy`, `ByteLengthQueuingStrategy`, `TextEncoderStream`, and `TextDecoderStream`, as globals and as `node:stream/web` exports, are constructor values with Node's prototypes, brand checks, and errors. Their underlying sources, sinks, and transformers, queuing strategies with a `size` function, readers and writers, `tee`, `pipeTo` and `pipeThrough` with their options and an abort signal, `ReadableStream.from`, and async iteration settle promises in Node's order, and the streams take own properties. `TextDecoderStream` decodes UTF-8; its other encodings are unsupported. diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index edb391f7d..7a7b61fd7 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/packages/compiler/ambient/scriptc-node-fallback.d.ts @@ -1115,9 +1115,9 @@ declare var ReadableStream: { from(iterable: string): ReadableStream; readonly prototype: ReadableStream; }; -/* The writable side of a TransformStream and its writer. The - * WritableStream constructor has no native lowering; writable streams come - * from `new TransformStream()`. */ +/* The writable side of a TransformStream and its writer. A program that + * constructs a WritableStream, or a TransformStream with a transformer, + * builds its streams with the web stream classes (scr_webstreams.c). */ interface WritableStreamDefaultWriter { readonly closed: Promise; readonly desiredSize: number | null; @@ -1133,14 +1133,25 @@ interface WritableStream { close(): Promise; getWriter(): WritableStreamDefaultWriter; } -/* The identity transform: chunks written to `writable` are read from - * `readable`, and a write completes when its chunk is read. */ +/* Without a transformer, the identity transform: chunks written to + * `writable` are read from `readable`, and a write completes when its chunk + * is read. */ interface TransformStream { readonly readable: ReadableStream; readonly writable: WritableStream; } +interface Transformer { + start?(controller: TransformStreamDefaultController): unknown; + transform?(chunk: I, controller: TransformStreamDefaultController): unknown; + flush?(controller: TransformStreamDefaultController): unknown; + cancel?(reason?: unknown): unknown; +} declare var TransformStream: { - new (): TransformStream; + new ( + transformer?: Transformer, + writableStrategy?: QueuingStrategy, + readableStrategy?: QueuingStrategy, + ): TransformStream; readonly prototype: TransformStream; }; /* The other web stream interfaces. JavaScript sources construct and read @@ -1158,8 +1169,14 @@ interface WritableStreamDefaultController { readonly signal: AbortSignal; error(reason?: unknown): void; } +interface UnderlyingSink { + start?(controller: WritableStreamDefaultController): unknown; + write?(chunk: W, controller: WritableStreamDefaultController): unknown; + close?(): unknown; + abort?(reason?: unknown): unknown; +} declare var WritableStream: { - new (sink?: object, strategy?: QueuingStrategy): WritableStream; + new (sink?: UnderlyingSink, strategy?: QueuingStrategy): WritableStream; readonly prototype: WritableStream; }; declare var WritableStreamDefaultWriter: { diff --git a/packages/compiler/src/compat/fetch-profile.ts b/packages/compiler/src/compat/fetch-profile.ts index 07f5f09c8..e723d2df6 100644 --- a/packages/compiler/src/compat/fetch-profile.ts +++ b/packages/compiler/src/compat/fetch-profile.ts @@ -155,6 +155,12 @@ const typedInterfaceUnsupported = "typed source has no compiler bridge for this const metadataExclusion = "WebIDL brand metadata is observable reflection, not an executable compatibility operation"; +/** ReadableStream methods a typed program reaches through the web stream + * classes (scr_webstreams.c): on a stream it constructs with them, and on + * the native stream through its web view (scr_fetch.c, sf_stream_web_view). + * A program that calls one links the classes. */ +export const READABLE_STREAM_WEB_VIEW_CALLS = ["pipeThrough", "pipeTo", "tee"] as const; + function buildFetchCompatProfile() { return { schemaVersion: 1, @@ -233,7 +239,7 @@ function buildFetchCompatProfile() { ], responseCalls: ["json", "text", "bytes", "clone"], readableStreamReads: ["locked"], - readableStreamCalls: ["cancel", "getReader"], + readableStreamCalls: ["cancel", "getReader", ...READABLE_STREAM_WEB_VIEW_CALLS], }, operations: [ { @@ -447,6 +453,13 @@ function buildFetchCompatProfile() { facets: ["webidl-conversion", "identity", "state-machine", "error-shape"], evidence: [generated("stream-traces"), fixture("static-stream")], }, + ...READABLE_STREAM_WEB_VIEW_CALLS.map((member): FetchCompatOperation => ({ + id: `stdlib.readable-stream.${member}`, + name: `ReadableStream.${member}`, + kind: "method", + facets: ["identity", "promise-settlement", "state-machine", "error-shape"], + evidence: [fixture("static-web-streams")], + })), { id: "stdlib.readable-stream-default-reader.closed", name: "ReadableStreamDefaultReader.closed", @@ -757,25 +770,15 @@ function buildFetchCompatProfile() { "getReader", "prototype", ), - ...["pipeThrough", "values"].map((member) => - unsupportedEntry( - `stdlib.readable-stream.${member}`, - "ReadableStream", - member, - "prototype", - member === "values" - ? "direct for-await consumption compiles statically; storing or driving the first-class iterator handle is unsupported" - : "the wider Web Streams graph is outside the native readable-stream slice", - ), + ...READABLE_STREAM_WEB_VIEW_CALLS.map((member) => + staticEntry(`stdlib.readable-stream.${member}`, "ReadableStream", member, "prototype"), ), - ...["pipeTo", "tee"].map((member) => - unsupportedEntry( - `stdlib.readable-stream.${member}`, - "ReadableStream", - member, - "prototype", - "this operation is outside the native readable-stream implementation", - ), + unsupportedEntry( + "stdlib.readable-stream.values", + "ReadableStream", + "values", + "prototype", + "direct for-await consumption compiles statically; storing or driving the first-class iterator handle is unsupported", ), unsupportedEntry( "stdlib.readable-stream.symbol.asyncIterator", @@ -954,7 +957,8 @@ export interface StaticFetchMembers { readableStreamReads: ReadonlySet; readableStreamCalls: ReadonlySet; /** ReadableStream prototype members the native stream does not - * implement, symbol-keyed ones as `[Symbol.name]`. */ + * implement itself, symbol-keyed ones as `[Symbol.name]`: the unsupported + * ones, and the calls it answers through its web view. */ readableStreamUnsupported: ReadonlySet; } @@ -979,8 +983,8 @@ export function staticFetchMembers(): StaticFetchMembers { ), readableStreamReads: new Set(profile.members.readableStreamReads), readableStreamCalls: new Set(profile.members.readableStreamCalls), - readableStreamUnsupported: new Set( - profile.inventory.entries + readableStreamUnsupported: new Set([ + ...profile.inventory.entries .filter( (entry) => entry.owner === "ReadableStream" && @@ -988,7 +992,8 @@ export function staticFetchMembers(): StaticFetchMembers { entry.status === "unsupported", ) .map((entry) => entry.member), - ), + ...READABLE_STREAM_WEB_VIEW_CALLS, + ]), }; return staticFetchMemberSets; } diff --git a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts index 765d3beae..67caec84a 100644 --- a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts +++ b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts @@ -1,4 +1,7 @@ -import { staticFetchMembers } from "../../../compat/fetch-profile.js"; +import { + READABLE_STREAM_WEB_VIEW_CALLS, + staticFetchMembers, +} from "../../../compat/fetch-profile.js"; import { numLit } from "../../../ir/build.js"; import { everyStmt } from "../../../ir/traverse.js"; import { @@ -16,6 +19,7 @@ import { constituentTypes } from "../../ts7/checker.js"; import { ENTRY_NAME, type Lowerer } from "../lowerer.js"; import { stdlibGlobalNameOf } from "../surfaces.js"; import { streamWebImportOf } from "./module-bindings.js"; +import { lowerDynObjectLiteral } from "../expressions/object-literals.js"; /* The web stream classes in JavaScript sources: the globals and the * node:stream/web exports, which are the same constructors. Each reference @@ -92,6 +96,63 @@ export function programNamesWebStreamClass(lowerer: Lowerer): boolean { return found; } +/** The classes a TypeScript source constructs through the web stream + * classes once it uses an operation the native streams lack. */ +const TYPED_CLASSES: ReadonlySet = new Set([ + "ReadableStream", + "WritableStream", + "TransformStream", +]); +const WEB_VIEW_CALLS: ReadonlySet = new Set(READABLE_STREAM_WEB_VIEW_CALLS); + +/** Whether a TypeScript source of the program uses a web stream operation + * the native streams lack: a ReadableStream tee(), pipeTo() or + * pipeThrough() call, a WritableStream constructor, or a TransformStream + * constructor with arguments (a transformer or queuing strategies). Such a + * program builds its ReadableStream, WritableStream and TransformStream + * values with the web stream classes, as JavaScript does; a program that + * uses only getReader(), cancel() and the identity transform keeps the + * native streams. */ +export function programUsesTypedWebStreams(lowerer: Lowerer): boolean { + if (lowerer.typedWebStreams !== undefined) return lowerer.typedWebStreams; + let found = false; + for (const file of lowerer.program.getSourceFiles()) { + if (file.isDeclarationFile || isJsSourceFile(file)) continue; + ts.walkPreorder(file, (node) => { + if ( + ts.isPropertyAccessExpression(node) && + WEB_VIEW_CALLS.has(node.name.text) && + isReadableStreamMember(lowerer, node) + ) + found = true; + else if (ts.isNewExpression(node)) { + const name = stdlibGlobalNameOf(lowerer, unparenthesized(node.expression)); + if ( + name === "WritableStream" || + (name === "TransformStream" && (node.arguments?.nodes.length ?? 0) > 0) + ) + found = true; + } + return found ? "stop" : undefined; + }); + if (found) break; + } + lowerer.typedWebStreams = found; + return found; +} + +/** The class a TypeScript `new` expression constructs through the web + * stream classes (programUsesTypedWebStreams), as its index in + * WEB_STREAM_CLASSES; null otherwise. */ +function typedWebStreamClassKindOf(lowerer: Lowerer, node: ts.Expression): number | null { + const expr = unparenthesized(node); + if (!ts.isIdentifier(expr) || isJsSourceFile(ts.getSourceFileOfNode(expr))) return null; + const name = streamWebImportOf(lowerer, expr) ?? stdlibGlobalNameOf(lowerer, expr); + if (name === null || !TYPED_CLASSES.has(name) || !programUsesTypedWebStreams(lowerer)) + return null; + return CLASS_NAMES.indexOf(name); +} + export function webStreamClassValue(kind: number, loc: SrcLoc): IrExpr { return { kind: "libCall", fn: "webstream.class", args: [numLit(kind, loc)], type: DYN, loc }; } @@ -102,13 +163,24 @@ export function lowerWebStreamClassRead(lowerer: Lowerer, node: ts.Expression): return kind === null ? null : webStreamClassValue(kind, locOf(node)); } -/** `new X(...args)` on one of the classes in JavaScript. */ +/** `new X(...args)` on one of the classes in JavaScript, and on + * ReadableStream, WritableStream and TransformStream in a TypeScript + * program that uses the web stream classes (programUsesTypedWebStreams). + * A TypeScript source's object literal argument (a source, sink, + * transformer or strategy) becomes a plain object with its methods. */ export function lowerWebStreamNew(lowerer: Lowerer, expr: ts.NewExpression): IrExpr | null { - const kind = webStreamClassKindOf(lowerer, expr.expression); + const typed = webStreamClassKindOf(lowerer, expr.expression) === null; + const kind = typed + ? typedWebStreamClassKindOf(lowerer, expr.expression) + : webStreamClassKindOf(lowerer, expr.expression); if (kind === null) return null; const nodes = expr.arguments?.nodes ?? []; if (nodes.some(ts.isSpreadElement)) return null; const loc = locOf(expr); + const argument = (arg: ts.Expression): IrExpr => + typed && ts.isObjectLiteralExpression(arg) + ? lowerDynObjectLiteral(lowerer, arg) + : lowerer.lowerExprExpecting(arg, DYN); return { kind: "libCall", fn: "dyn.construct", @@ -116,7 +188,7 @@ export function lowerWebStreamNew(lowerer: Lowerer, expr: ts.NewExpression): IrE webStreamClassValue(kind, loc), { kind: "dynArrLit", - elems: nodes.map((arg) => lowerer.lowerExprExpecting(arg, DYN)), + elems: nodes.map(argument), type: DYN, loc, }, @@ -285,6 +357,7 @@ function programLinksFetch(functions: IrFunction[]): boolean { * count in a program that links the fetch unit, since the value can be a * fetch body. */ function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction[]): boolean { + if (programUsesTypedWebStreams(lowerer)) return true; const members = staticFetchMembers().readableStreamUnsupported; const iterates = members.has("[Symbol.asyncIterator]"); let untyped = false; @@ -310,7 +383,9 @@ function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction } /** Links the web stream classes into a program whose JavaScript uses a - * ReadableStream member the fetch unit's own stream lacks: a fetch body's + * ReadableStream member the fetch unit's own stream lacks, or whose + * TypeScript uses the web stream classes (programUsesTypedWebStreams): a + * fetch body's or a native stream's * handle stands for a web ReadableStream once such a member is used * (scr_fetch.c, sf_stream_web_view). %main installs the classes first. A * program that names one of the classes links them already. */ diff --git a/packages/compiler/src/frontend/lowering/lower-stdlib.ts b/packages/compiler/src/frontend/lowering/lower-stdlib.ts index 44f2b7370..f1db4bcf1 100644 --- a/packages/compiler/src/frontend/lowering/lower-stdlib.ts +++ b/packages/compiler/src/frontend/lowering/lower-stdlib.ts @@ -1425,7 +1425,7 @@ export function fenceStaticReadableStreamMember( lowerer.noLowering( `ReadableStream.${member} in a static build`, access, - "the native static ReadableStream surface is locked plus cancel() and getReader(); values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while stored iterator handles remain unsupported", + "ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported", sym, ); } @@ -1965,8 +1965,9 @@ export function lowerStaticReadableStreamNew( /** `new TransformStream()`: the identity transform, a native handle whose * readable side is a native ReadableStream and whose writable side is the - * runtime's WritableStream. A transformer or queuing strategies have no - * native lowering yet and keep the SC2020 refusal. */ + * runtime's WritableStream. A TransformStream with a transformer or + * queuing strategies is built with the web stream classes instead + * (web-streams-dyn.ts, lowerWebStreamNew). */ export function lowerStaticTransformStreamNew( lowerer: Lowerer, expr: ts.NewExpression, diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts index 6ae1ea9f5..c3a5ec42f 100644 --- a/packages/compiler/src/frontend/lowering/lowerer.ts +++ b/packages/compiler/src/frontend/lowering/lowerer.ts @@ -3783,6 +3783,9 @@ export class Lowerer { /** web-streams-dyn.ts's answer: whether a JavaScript source names a web * stream class; undefined until asked. */ namesWebStreamClass: boolean | undefined = undefined; + /** web-streams-dyn.ts's answer: whether a TypeScript source uses a web + * stream operation the native streams lack; undefined until asked. */ + typedWebStreams: boolean | undefined = undefined; /** The first lifted function of each name (liftedFnNamed), for the * liftedFns indexed so far. liftedFns only grows. */ private readonly liftedFnsByName = new Map(); @@ -4413,6 +4416,7 @@ export class Lowerer { "sloppyThisGlobalReads", "globalLookupNames", "namesWebStreamClass", + "typedWebStreams", "timeValueDateDecisions", "timeValueDateWrites", "dynamicOverloadAnswers", diff --git a/packages/compiler/src/frontend/type-mapper.ts b/packages/compiler/src/frontend/type-mapper.ts index 36dae9b10..12aa672f4 100644 --- a/packages/compiler/src/frontend/type-mapper.ts +++ b/packages/compiler/src/frontend/type-mapper.ts @@ -220,8 +220,10 @@ export const FETCH_AMBIENT_TYPES = [ "ReadableStreamReadValueResult", "ReadableStreamReadDoneResult", "TransformStream", + "TransformStreamDefaultController", "WritableStream", "WritableStreamDefaultWriter", + "WritableStreamDefaultController", "Blob", "File", "BlobPropertyBag", diff --git a/packages/compiler/surface-manifest.json b/packages/compiler/surface-manifest.json index ea8295406..480a18409 100644 --- a/packages/compiler/surface-manifest.json +++ b/packages/compiler/surface-manifest.json @@ -5039,17 +5039,15 @@ "id": "stdlib.readable-stream.pipeThrough", "kind": "stdlib", "name": "ReadableStream.pipeThrough", - "status": "unsupported", - "code": "SC2020", - "note": "Node 24.22.0 / Undici 7.29.1; the wider Web Streams graph is outside the native readable-stream slice" + "status": "static", + "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams" }, { "id": "stdlib.readable-stream.pipeTo", "kind": "stdlib", "name": "ReadableStream.pipeTo", - "status": "unsupported", - "code": "SC2020", - "note": "Node 24.22.0 / Undici 7.29.1; this operation is outside the native readable-stream implementation" + "status": "static", + "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams" }, { "id": "stdlib.readable-stream.symbol.asyncIterator", @@ -5063,9 +5061,8 @@ "id": "stdlib.readable-stream.tee", "kind": "stdlib", "name": "ReadableStream.tee", - "status": "unsupported", - "code": "SC2020", - "note": "Node 24.22.0 / Undici 7.29.1; this operation is outside the native readable-stream implementation" + "status": "static", + "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams" }, { "id": "stdlib.readable-stream.values", diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-pipes.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-pipes.ts.json new file mode 100644 index 000000000..8455ba03b --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-pipes.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/web-streams-typed-pipes.ts" + ], + "diags": [] +} diff --git a/tests/corpus/web-streams-typed-pipes.ts b/tests/corpus/web-streams-typed-pipes.ts new file mode 100644 index 000000000..ab5564baa --- /dev/null +++ b/tests/corpus/web-streams-typed-pipes.ts @@ -0,0 +1,158 @@ +// ReadableStream tee(), pipeTo() and pipeThrough() in TypeScript, with +// WritableStream sinks and TransformStream transformers: chunk order, close +// and flush, cancel through a tee branch, the pipe options, Node's +// TypeErrors for locked streams, error propagation through a pipe, and a +// Response body piped through a transform. + +const describe = (e: unknown): string => `${(e as Error).name}: ${(e as Error).message}`; + +function source(chunks: string[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }); +} + +async function drain(stream: ReadableStream): Promise { + const out: T[] = []; + const reader = stream.getReader(); + for (;;) { + const result = await reader.read(); + if (result.done) break; + out.push(result.value); + } + return out; +} + +async function tees(): Promise { + const [left, right] = source(["a", "b", "c"]).tee(); + console.log("tee", (await drain(left)).join(""), (await drain(right)).join("")); + + const log: string[] = []; + const watched = new ReadableStream({ + start(controller) { + controller.enqueue(1); + controller.enqueue(2); + }, + cancel(reason) { + log.push(`source cancel ${String(reason)}`); + }, + }); + const [a, b] = watched.tee(); + // The source is cancelled once both branches are, with both reasons. + const leftDone = a.cancel("left done").then(() => log.push("left cancelled")); + log.push("left requested"); + await Promise.all([leftDone, b.cancel("right done").then(() => log.push("right cancelled"))]); + console.log("tee cancel", log.join(" | "), watched.locked); + + const locked = source(["x"]); + locked.getReader(); + try { + locked.tee(); + } catch (e) { + console.log("tee locked", describe(e)); + } +} + +async function pipes(): Promise { + const written: string[] = []; + const sink = new WritableStream({ + write(chunk) { + written.push(chunk); + }, + close() { + written.push(""); + }, + }); + await source(["one", "two"]).pipeTo(sink); + console.log("pipeTo", written.join(","), sink.locked); + + const kept: string[] = []; + const open = new WritableStream({ + write(chunk) { + kept.push(chunk); + }, + close() { + kept.push(""); + }, + }); + await source(["p"]).pipeTo(open, { preventClose: true }); + const writer = open.getWriter(); + await writer.write("after"); + await writer.close(); + console.log("preventClose", kept.join(",")); + + const failing = new ReadableStream({ + start(controller) { + controller.enqueue("ok"); + controller.error(new RangeError("source failed")); + }, + }); + const aborted: string[] = []; + try { + await failing.pipeTo( + new WritableStream({ + write(chunk) { + aborted.push(chunk); + }, + abort(reason) { + aborted.push(`abort ${describe(reason)}`); + }, + }), + ); + } catch (e) { + console.log("pipeTo error", describe(e), aborted.join(" | ")); + } + + const busy = source(["q"]); + busy.getReader(); + try { + await busy.pipeTo(new WritableStream()); + } catch (e) { + console.log("pipeTo locked", describe(e)); + } +} + +async function transforms(): Promise { + const upper = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk.toUpperCase()); + }, + flush(controller) { + controller.enqueue("!"); + }, + }); + console.log("pipeThrough", (await drain(source(["ab", "cd"]).pipeThrough(upper))).join("")); + + const lengths = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk.length); + }, + }); + console.log("lengths", (await drain(source(["a", "bbb", "cc"]).pipeThrough(lengths))).join(" ")); + + const identity = new TransformStream(); + console.log("identity", (await drain(source(["i", "j"]).pipeThrough(identity))).join("")); + + const body = new Response("fetch body").body!; + const decoded = body.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + controller.enqueue(new TextDecoder().decode(chunk)); + }, + }), + ); + console.log("response body", (await drain(decoded)).join("")); + + const [first, second] = new Response("split").body!.tee(); + const decoder = new TextDecoder(); + const text = async (s: ReadableStream): Promise => + (await drain(s)).map((chunk) => decoder.decode(chunk)).join(""); + console.log("response tee", await text(first), await text(second)); +} + +await tees(); +await pipes(); +await transforms(); diff --git a/tests/fixtures/fetch/static-web-streams/main.mts b/tests/fixtures/fetch/static-web-streams/main.mts new file mode 100644 index 000000000..e8e171d66 --- /dev/null +++ b/tests/fixtures/fetch/static-web-streams/main.mts @@ -0,0 +1,96 @@ +// ReadableStream tee(), pipeTo() and pipeThrough() in a static build: a +// fetched body (whole and chunked) tee'd, piped into a WritableStream and +// through a TransformStream, constructed streams through a transformer, +// and Node's TypeErrors for locked streams and errors through a pipe. +const base = process.argv[2]!; +const describe = (e: unknown): string => (e instanceof Error ? `${e.name}: ${e.message}` : String(e)); +const decoder = new TextDecoder(); + +async function collect(stream: ReadableStream): Promise { + const reader = stream.getReader(); + let text = ""; + for (;;) { + const result = await reader.read(); + if (result.done) break; + text += decoder.decode(result.value, { stream: true }); + } + return text + decoder.decode(); +} + +const fetched = await fetch(`${base}/text`); +const [left, right] = fetched.body!.tee(); +console.log("tee", await collect(left), await collect(right), fetched.bodyUsed); + +const chunked = await fetch(`${base}/chunked`); +const parts: string[] = []; +await chunked.body!.pipeTo( + new WritableStream({ + write(chunk) { + parts.push(decoder.decode(chunk, { stream: true })); + }, + close() { + parts.push(""); + }, + }), +); +console.log("pipeTo", parts.join("").length > 0, parts[parts.length - 1], chunked.bodyUsed); + +const upper = await fetch(`${base}/text`); +const shouted = upper.body!.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + controller.enqueue(new TextEncoder().encode(decoder.decode(chunk).toUpperCase())); + }, + }), +); +console.log("pipeThrough", await collect(shouted)); + +const words = new ReadableStream({ + start(controller) { + controller.enqueue("alpha"); + controller.enqueue("beta"); + controller.close(); + }, +}); +const counted = words.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk.length); + }, + flush(controller) { + controller.enqueue(-1); + }, + }), +); +const counts: number[] = []; +const reader = counted.getReader(); +for (;;) { + const result = await reader.read(); + if (result.done) break; + counts.push(result.value); +} +console.log("constructed", counts.join(" ")); + +const locked = await fetch(`${base}/text`); +locked.body!.getReader(); +try { + locked.body!.tee(); +} catch (e) { + console.log("tee locked", describe(e)); +} +try { + await locked.body!.pipeTo(new WritableStream()); +} catch (e) { + console.log("pipeTo locked", describe(e)); +} + +const failing = new ReadableStream({ + start(controller) { + controller.error(new TypeError("source failed")); + }, +}); +try { + await failing.pipeTo(new WritableStream()); +} catch (e) { + console.log("pipeTo error", describe(e)); +} diff --git a/tests/harness/__snapshots__/node-types-fenced.txt b/tests/harness/__snapshots__/node-types-fenced.txt index ada6231d4..073a031e5 100644 --- a/tests/harness/__snapshots__/node-types-fenced.txt +++ b/tests/harness/__snapshots__/node-types-fenced.txt @@ -142,41 +142,23 @@ fenced.ts:80:3 - error SC2020: 'RequestInit option 'integrity' in a static build hint: the compiler does not preserve this RequestInit member's conversion or transport behavior -fenced.ts:84:3 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet - - 83 | const body = (await fetch(url)).body!; - 84 | body.tee(); - | ^~~~~~~~ - 85 | const tee = body.tee; - - hint: this operation is outside the native readable-stream implementation - -fenced.ts:85:15 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet +fenced.ts:85:15 - error SC2020: 'ReadableStream.tee in a static build' is typed by @types/node but has no scriptc lowering yet 84 | body.tee(); 85 | const tee = body.tee; | ^~~~~~~~ 86 | body["tee"](); - hint: this operation is outside the native readable-stream implementation + hint: ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported -fenced.ts:86:3 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet - - 85 | const tee = body.tee; - 86 | body["tee"](); - | ^~~~~~~~~~~ - 87 | const bracketTee = body["tee"]; - - hint: this operation is outside the native readable-stream implementation - -fenced.ts:87:22 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet +fenced.ts:87:22 - error SC2020: 'ReadableStream.tee in a static build' is typed by @types/node but has no scriptc lowering yet 86 | body["tee"](); 87 | const bracketTee = body["tee"]; | ^~~~~~~~~~~ 88 | } - hint: this operation is outside the native readable-stream implementation + hint: ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported fenced.ts:93:23 - error SC2020: 'util.callbackify into a callback whose error parameter excludes null' is part of the standard library types but has no scriptc lowering yet diff --git a/tests/harness/corpus-costs/web-streams-typed-pipes.ts.json b/tests/harness/corpus-costs/web-streams-typed-pipes.ts.json new file mode 100644 index 000000000..5cce6d4a6 --- /dev/null +++ b/tests/harness/corpus-costs/web-streams-typed-pipes.ts.json @@ -0,0 +1 @@ +{ "plain": 3000, "san": 6000 } diff --git a/tests/harness/fetch.test.ts b/tests/harness/fetch.test.ts index 1df7965f6..acd93af81 100644 --- a/tests/harness/fetch.test.ts +++ b/tests/harness/fetch.test.ts @@ -165,6 +165,7 @@ describe(`static fetch differential${sanitize ? " (sanitized)" : ""}`, () => { "static-dispatch-throw", "static-response-json", "static-clone", + "static-web-streams", ] as const; test.for( staticCases.flatMap((name) => (["llvm"] as const).map((backend) => [name, backend] as const)), diff --git a/tests/harness/surface-manifest.test.ts b/tests/harness/surface-manifest.test.ts index 710cc529b..c4650151f 100644 --- a/tests/harness/surface-manifest.test.ts +++ b/tests/harness/surface-manifest.test.ts @@ -450,7 +450,7 @@ test("unimplemented Response and ReadableStream calls stay fenced", () => { " void value.blob(); void value.formData();\n" + "}\n" + "function stream(value: ReadableStream): void {\n" + - " void value.tee(); void value.pipeTo(null as never);\n" + + " void value.values(); void value[Symbol.asyncIterator]();\n" + "}\n" + "void response; void stream;\n", }); From 64f9481fb8a15bec461f2b5f3b94af2b398006d8 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Thu, 29 Oct 2026 04:41:05 -0500 Subject: [PATCH 2/2] Link the web stream classes for an element-access tee, pipeTo or pipeThrough The scan that decides whether a program uses a web stream operation the native streams lack only matched property accesses, so a TypeScript program whose one such operation was `body["tee"]()` kept the native streams and threw a TypeError at the call. Element accesses with a string literal key now count the same way, in the TypeScript scan and in the JavaScript one. Adds the corpus program web-streams-typed-element-tee.ts, with its corpus cost and order baseline. --- .../lowering/builtins/web-streams-dyn.ts | 39 +++++++++++++------ .../web-streams-typed-element-tee.ts.json | 6 +++ tests/corpus/web-streams-typed-element-tee.ts | 23 +++++++++++ .../web-streams-typed-element-tee.ts.json | 1 + 4 files changed, 58 insertions(+), 11 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-element-tee.ts.json create mode 100644 tests/corpus/web-streams-typed-element-tee.ts create mode 100644 tests/harness/corpus-costs/web-streams-typed-element-tee.ts.json diff --git a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts index 67caec84a..809d5bb03 100644 --- a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts +++ b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts @@ -119,11 +119,8 @@ export function programUsesTypedWebStreams(lowerer: Lowerer): boolean { for (const file of lowerer.program.getSourceFiles()) { if (file.isDeclarationFile || isJsSourceFile(file)) continue; ts.walkPreorder(file, (node) => { - if ( - ts.isPropertyAccessExpression(node) && - WEB_VIEW_CALLS.has(node.name.text) && - isReadableStreamMember(lowerer, node) - ) + const member = staticMemberName(node); + if (member !== null && WEB_VIEW_CALLS.has(member) && isReadableStreamMember(lowerer, node)) found = true; else if (ts.isNewExpression(node)) { const name = stdlibGlobalNameOf(lowerer, unparenthesized(node.expression)); @@ -316,10 +313,29 @@ function isReadableStreamType(lowerer: Lowerer, type: ts.Type): boolean { ); } -/** Whether a member access resolves to a member of the stdlib - * ReadableStream interface. */ -function isReadableStreamMember(lowerer: Lowerer, access: ts.PropertyAccessExpression): boolean { - const member = lowerer.checker.getSymbolAtLocation(access.name); +type MemberAccess = ts.PropertyAccessExpression | ts.ElementAccessExpression; + +/** The member a property access names, or an element access with a string + * literal key (`body.tee` and `body["tee"]`); null for any other node. */ +function staticMemberName(node: ts.Node): string | null { + if (ts.isPropertyAccessExpression(node)) return node.name.text; + if (ts.isElementAccessExpression(node)) { + const key = node.argumentExpression; + if (ts.isStringLiteral(key) || ts.isNoSubstitutionTemplateLiteral(key)) return key.text; + } + return null; +} + +/** Whether a member access (staticMemberName) resolves to a member of the + * stdlib ReadableStream interface. */ +function isReadableStreamMember(lowerer: Lowerer, access: ts.Node): boolean { + const key = ts.isPropertyAccessExpression(access) + ? access.name + : ts.isElementAccessExpression(access) + ? access.argumentExpression + : null; + if (key === null) return false; + const member = lowerer.checker.getSymbolAtLocation(key); if (member === undefined || !lowerer.isStdlibSymbol(member)) return false; let owner: ts.Node | undefined = member.declarations[0]?.parent; while (owner !== undefined && !ts.isInterfaceDeclaration(owner)) owner = owner.parent; @@ -366,9 +382,10 @@ function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction let found = false; ts.walkPreorder(file, (node) => { let receiver: ts.Type | null = null; - if (ts.isPropertyAccessExpression(node) && members.has(node.name.text)) { + const member = staticMemberName(node); + if (member !== null && members.has(member)) { if (isReadableStreamMember(lowerer, node)) found = true; - else receiver = lowerer.typeOf(node.expression); + else receiver = lowerer.typeOf((node as MemberAccess).expression); } else if (iterates && ts.isForOfStatement(node) && node.awaitModifier !== undefined) { receiver = lowerer.typeOf(node.expression); if (isReadableStreamType(lowerer, receiver)) found = true; diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-element-tee.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-element-tee.ts.json new file mode 100644 index 000000000..61b037166 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-element-tee.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/web-streams-typed-element-tee.ts" + ], + "diags": [] +} diff --git a/tests/corpus/web-streams-typed-element-tee.ts b/tests/corpus/web-streams-typed-element-tee.ts new file mode 100644 index 000000000..744ff6bb3 --- /dev/null +++ b/tests/corpus/web-streams-typed-element-tee.ts @@ -0,0 +1,23 @@ +// ReadableStream tee() reached through an element access with a string key +// is the program's only web stream operation: the program still links the +// web stream classes, so a Response body splits into two branches. + +async function text(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let out = ""; + for (;;) { + const result = await reader.read(); + if (result.done) break; + out += decoder.decode(result.value, { stream: true }); + } + return out + decoder.decode(); +} + +async function main(): Promise { + const body = new Response("split me").body!; + const [left, right] = body["tee"](); + console.log("element tee", await text(left), "|", await text(right), body.locked); +} + +void main(); diff --git a/tests/harness/corpus-costs/web-streams-typed-element-tee.ts.json b/tests/harness/corpus-costs/web-streams-typed-element-tee.ts.json new file mode 100644 index 000000000..5cce6d4a6 --- /dev/null +++ b/tests/harness/corpus-costs/web-streams-typed-element-tee.ts.json @@ -0,0 +1 @@ +{ "plain": 3000, "san": 6000 }