From 17efed329fcc2a8651b65d41c0db03b9097163a7 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Wed, 28 Oct 2026 03:44:57 -0500 Subject: [PATCH 1/2] Give JavaScript fetch bodies the members of a web ReadableStream In JavaScript, a Response or Request body answered getReader, cancel and locked, and every other ReadableStream member was undefined: tee, pipeTo, pipeThrough, values and for await threw TypeErrors where Node's body is a ReadableStream. In programs that link the web stream classes, the body's handle now stands for a ReadableStream once any other member is used (sf_stream_web_view): a stream with a zero high-water mark whose native source reads the body (scr_webstream_readable_over). From then on every member goes to it, the class's method values and getters run on it through the alias's view (scr_dyn_handle_class_receiver), and text(), clone(), new Response(body) and fetch uploads read the body through it, so its lock and bodyUsed follow it as in Node. A body locked when the view is made keeps the view locked until the lock is released. Once only the view's source holds the body, the body lets the view go. The compiler links the classes into a program whose JavaScript reads a ReadableStream member that the fetch profile lists as unsupported by the native stream, or iterates a ReadableStream with for await, and that names none of the classes: %main calls webstream.install first. Programs without web streams build to the same binaries as before. Adds corpus programs fetch-body-web-stream-linked.mjs, fetch-body-web-stream-unlinked.mjs and fetch-body-web-stream-network.mjs, with their corpus costs and preflight/order baselines. --- .../builtin-apis/streams-and-readline.mdx | 3 + .../compiler/src/backend/llvm/lib-dispatch.ts | 2 +- .../compiler/src/backend/llvm/lib-shared.ts | 1 + packages/compiler/src/compat/fetch-profile.ts | 13 + .../lowering/builtins/web-streams-dyn.ts | 82 +++++- .../compiler/src/frontend/lowering/lowerer.ts | 7 +- .../compiler/src/ir/builtin-signatures.ts | 1 + packages/compiler/src/ir/runtime-features.ts | 4 + .../fetch-body-web-stream-linked.mjs.json | 6 + .../fetch-body-web-stream-network.mjs.json | 6 + .../fetch-body-web-stream-unlinked.mjs.json | 6 + packages/runtime/src/scr_dyn.h | 10 + packages/runtime/src/scr_dyn_handle_class.c | 29 ++- packages/runtime/src/scr_fetch.c | 244 +++++++++++++++++- packages/runtime/src/scr_generator.h | 18 ++ packages/runtime/src/scr_json.c | 20 +- packages/runtime/src/scr_webstreams.c | 92 +++++++ tests/corpus/fetch-body-web-stream-linked.mjs | 93 +++++++ .../corpus/fetch-body-web-stream-network.mjs | 26 ++ .../corpus/fetch-body-web-stream-unlinked.mjs | 25 ++ .../fetch-body-web-stream-linked.mjs.json | 1 + .../fetch-body-web-stream-network.mjs.json | 1 + .../fetch-body-web-stream-unlinked.mjs.json | 1 + 23 files changed, 671 insertions(+), 20 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json create mode 100644 tests/corpus/fetch-body-web-stream-linked.mjs create mode 100644 tests/corpus/fetch-body-web-stream-network.mjs create mode 100644 tests/corpus/fetch-body-web-stream-unlinked.mjs create mode 100644 tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json create mode 100644 tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json create mode 100644 tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index f3883fe8d..52bf534e8 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -65,4 +65,7 @@ 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. A JavaScript `ReadableStream` given as a `Response` or `Request` body, or as `fetch`'s body, is read through a stream of its own: `response.body` is that stream rather than the one passed in, and the stream passed in is locked from the start. + In JavaScript, a `Response` or `Request` body has the members of a `ReadableStream`, including `tee`, `pipeTo`, `pipeThrough`, `values`, and `for await`, and its lock and `bodyUsed` follow them. + After one of those members is used, a body given to a new `Response` or `Request` is locked from the start, as a JavaScript `ReadableStream` is. + In a program that names none of the web stream classes, those members are available when the program's JavaScript names one of them or iterates a body with `for await`; reached only through a computed key, they are not. In JavaScript, a byte stream (`type: 'bytes'`), a BYOB reader, a class that extends one of these classes, `Object.getPrototypeOf` on these values, their use as `WeakMap` and `WeakSet` keys, and `ReadableStream.from` of a function that is not iterable are unsupported. diff --git a/packages/compiler/src/backend/llvm/lib-dispatch.ts b/packages/compiler/src/backend/llvm/lib-dispatch.ts index 823853ec8..57a09d85d 100644 --- a/packages/compiler/src/backend/llvm/lib-dispatch.ts +++ b/packages/compiler/src/backend/llvm/lib-dispatch.ts @@ -1274,7 +1274,7 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue { case "webstream": // A stream's start, pull and cancel steps run as promise jobs, so // a program with the classes runs its event loop. - if (e.fn === "webstream.class") host.usesTimers = true; + if (e.fn === "webstream.class" || e.fn === "webstream.install") host.usesTimers = true; return host.emitGenericLibCall(e); case "test": return emitTestLibCall(host, e); diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index b6bba291f..25578b815 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -120,6 +120,7 @@ export const LIB_FN_SYMS: Record = { "util.throttle": "scr_util_throttle", "webstream.class": "scr_webstream_class", "webstream.instanceOf": "scr_webstream_instanceof", + "webstream.install": "scr_webstream_install", "mime.typeNew": "scr_mime_type_new", "mime.typeParse": "scr_mime_type_parse", "mime.paramsNew": "scr_mime_params_new", diff --git a/packages/compiler/src/compat/fetch-profile.ts b/packages/compiler/src/compat/fetch-profile.ts index 03d261aff..83d809ee1 100644 --- a/packages/compiler/src/compat/fetch-profile.ts +++ b/packages/compiler/src/compat/fetch-profile.ts @@ -938,6 +938,9 @@ export interface StaticFetchMembers { headersCalls: ReadonlySet; readableStreamReads: ReadonlySet; readableStreamCalls: ReadonlySet; + /** ReadableStream prototype members the native stream does not + * implement, symbol-keyed ones as `[Symbol.name]`. */ + readableStreamUnsupported: ReadonlySet; } let staticFetchMemberSets: StaticFetchMembers | undefined; @@ -961,6 +964,16 @@ export function staticFetchMembers(): StaticFetchMembers { ), readableStreamReads: new Set(profile.members.readableStreamReads), readableStreamCalls: new Set(profile.members.readableStreamCalls), + readableStreamUnsupported: new Set( + profile.inventory.entries + .filter( + (entry) => + entry.owner === "ReadableStream" && + (entry.placement === "prototype" || entry.placement === "prototype-symbol") && + entry.status === "unsupported", + ) + .map((entry) => entry.member), + ), }; 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 24f603198..f18f21e16 100644 --- a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts +++ b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts @@ -1,9 +1,18 @@ +import { staticFetchMembers } from "../../../compat/fetch-profile.js"; import { numLit } from "../../../ir/build.js"; -import { BOOL, DYN, STRING, type IrExpr, type SrcLoc } from "../../../ir/ir.js"; +import { + BOOL, + DYN, + STRING, + VOID, + type IrExpr, + type IrFunction, + type SrcLoc, +} from "../../../ir/ir.js"; import * as ts from "../../ts7/adapter.js"; import { isJsSourceFile, locOf } from "../../program.js"; import { constituentTypes } from "../../ts7/checker.js"; -import type { Lowerer } from "../lowerer.js"; +import { ENTRY_NAME, type Lowerer } from "../lowerer.js"; import { stdlibGlobalNameOf } from "../surfaces.js"; import { streamWebImportOf } from "./module-bindings.js"; @@ -221,3 +230,72 @@ export function isWebStreamValueInJavaScript(lowerer: Lowerer, node: ts.Expressi return true; return typeInvolvesWebStream(lowerer, lowerer.typeOf(node)); } + +/** Whether a checker type is the stdlib ReadableStream, or a union with it. */ +function isReadableStreamType(lowerer: Lowerer, type: ts.Type): boolean { + if ((type.flags & ts.TypeFlags.Union) !== 0) + return constituentTypes(type).some((t) => isReadableStreamType(lowerer, t)); + const symbol = type.getSymbol(); + return ( + symbol !== undefined && + ts.symbolName(symbol) === "ReadableStream" && + lowerer.isStdlibSymbol(symbol) + ); +} + +/** 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); + 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; + return owner !== undefined && owner.name.text === "ReadableStream"; +} + +/** Whether a JavaScript source of the program uses a ReadableStream member + * that the fetch unit's own stream does not implement (the fetch profile's + * readableStreamUnsupported): a member access the checker resolves to the + * interface, or `for await` over a ReadableStream value. */ +function programUsesWebStreamOnlyMembers(lowerer: Lowerer): boolean { + const members = staticFetchMembers().readableStreamUnsupported; + const iterates = members.has("[Symbol.asyncIterator]"); + for (const file of lowerer.program.getSourceFiles()) { + if (file.isDeclarationFile || !isJsSourceFile(file)) continue; + let found = false; + ts.walkPreorder(file, (node) => { + if ( + (ts.isPropertyAccessExpression(node) && + members.has(node.name.text) && + isReadableStreamMember(lowerer, node)) || + (iterates && + ts.isForOfStatement(node) && + node.awaitModifier !== undefined && + isReadableStreamType(lowerer, lowerer.typeOf(node.expression))) + ) { + found = true; + return "stop"; + } + return undefined; + }); + if (found) return true; + } + return false; +} + +/** 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 + * 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. */ +export function installFetchBodyWebStreams(lowerer: Lowerer, functions: IrFunction[]): void { + if (programNamesWebStreamClass(lowerer) || !programUsesWebStreamOnlyMembers(lowerer)) return; + const main = functions.find((fn) => fn.name === ENTRY_NAME); + if (!main) return; + const loc = main.loc; + main.body.unshift({ + kind: "exprStmt", + expr: { kind: "libCall", fn: "webstream.install", args: [], type: VOID, loc }, + loc, + }); +} diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts index 2c3864964..e6b6a4653 100644 --- a/packages/compiler/src/frontend/lowering/lowerer.ts +++ b/packages/compiler/src/frontend/lowering/lowerer.ts @@ -118,7 +118,11 @@ import { fenceWebCryptoMember, lowerWebCryptoCall } from "./builtins/webcrypto.j import { valueStreamOf } from "./builtins/hash-stream.js"; import { lowerNetAddressCall, lowerNetAddressNew } from "./builtins/net-address.js"; import { lowerMimeCall, lowerMimeNew } from "./builtins/mime.js"; -import { lowerWebStreamNew, lowerWebStreamStaticCall } from "./builtins/web-streams-dyn.js"; +import { + installFetchBodyWebStreams, + lowerWebStreamNew, + lowerWebStreamStaticCall, +} from "./builtins/web-streams-dyn.js"; import { dynUndefinedExpr, nodeThrowExpr, numLit, varRef } from "../../ir/build.js"; import { iteratorSlotAdapter } from "./iterator-adapters.js"; import { iterableSlotValue } from "./iterable-values.js"; @@ -9459,6 +9463,7 @@ export class Lowerer { guardFrozenWrites(this, functions); applyHandleProperties(this, functions); installBufferValue(functions); + installFetchBodyWebStreams(this, functions); // Globals typed by a class that never REGISTERED (a JS class whose // collection fenced — Symbol-keyed fields, an unsupported base): the diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index b017e4e3d..35b272dd2 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -1209,6 +1209,7 @@ function buildLibFnSignatures() { "util.throttle": { argTypes: [DYN, DYN, DYN, DYN], result: DYN }, "webstream.class": { argTypes: [F64], result: DYN }, "webstream.instanceOf": { argTypes: [DYN, F64], result: BOOL }, + "webstream.install": { argTypes: [], result: VOID }, "mime.typeNew": { argTypes: [DYN, DYN], result: DYN }, "mime.typeParse": { argTypes: [DYN], result: DYN }, "mime.paramsNew": { argTypes: [], result: DYN }, diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 04392fe48..58fce50b6 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -83,6 +83,7 @@ const DYN_ASYNC_LIB_FNS: ReadonlySet = new Set([ // Web streams settle their promises with dyn values. "webstream.class", "webstream.instanceOf", + "webstream.install", "dyn.promiseAll", "dyn.promiseAllSettled", "dyn.promiseAny", @@ -390,6 +391,7 @@ const SYMBOL_LIB_FNS: ReadonlySet = new Set([ // Web streams answer async iteration through Symbol.asyncIterator. "webstream.class", "webstream.instanceOf", + "webstream.install", // EventEmitter's kShapeMode and kCapture properties ...BUILTIN_CLASS_LIB_FNS, "dyn.usingDispose", @@ -954,6 +956,7 @@ const RUNTIME_FEATURE_RULES = { // scr_webstreams.c calls the iterators ReadableStream.from reads. "webstream.class", "webstream.instanceOf", + "webstream.install", ], // Constant-key reads of Object.prototype member names answer the // inherited member through scr_dyn_invoke.c. @@ -1076,6 +1079,7 @@ const RUNTIME_FEATURE_RULES = { "global.native", // ERR_ARG_NOT_ITERABLE renders the value as util.format's %s does. "webstream.class", + "webstream.install", "util.styleText", "crypto.cipherNew", "webcrypto.encrypt", diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json new file mode 100644 index 000000000..eaed983e9 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-linked.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json new file mode 100644 index 000000000..44d798239 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-network.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json new file mode 100644 index 000000000..1b37cd0c0 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-unlinked.mjs" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_dyn.h b/packages/runtime/src/scr_dyn.h index fafa16eda..c2aed501e 100644 --- a/packages/runtime/src/scr_dyn.h +++ b/packages/runtime/src/scr_dyn.h @@ -2084,6 +2084,16 @@ ScrDyn *scr_dyn_handle_class_prototype(ScrDynHandleTag tag); /* Handles of `tag` (another unit's native representation of the same * interface) are instances of the class of `klass_tag`. */ void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag); +/* The instance of the class a handle of an alias tag stands for, which + * the class's methods and getters run on (+1, or NULL with the exception + * pending). Without one, they throw their ERR_INVALID_THIS TypeError on + * the alias's handles. */ +void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle)); +/* The receiver a method or getter of the class of `klass_tag` runs on for + * `self`: true with *out set to self, or to the instance an alias handle + * stands for (+1, or NULL with the exception pending); false when self is + * neither. */ +bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out); /* The class tag a handle is an instance of, or SCR_DYNH_CAPACITY for a * handle of no class. */ ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle); diff --git a/packages/runtime/src/scr_dyn_handle_class.c b/packages/runtime/src/scr_dyn_handle_class.c index cf2a04f76..30e2b5500 100644 --- a/packages/runtime/src/scr_dyn_handle_class.c +++ b/packages/runtime/src/scr_dyn_handle_class.c @@ -20,6 +20,8 @@ static SCR_TL const ScrDynHandleClass *hc_classes[SCR_DYNH_CAPACITY]; /* The class tag of another tag's handles + 1 (scr_dyn_handle_class_alias); * 0 = none. */ static SCR_TL uint16_t hc_aliases[SCR_DYNH_CAPACITY]; +/* scr_dyn_handle_class_alias_view, per alias tag. */ +static SCR_TL ScrDyn *(*hc_alias_views[SCR_DYNH_CAPACITY])(ScrDyn *handle); static SCR_TL ScrDyn *hc_ctors[SCR_DYNH_CAPACITY]; static SCR_TL ScrDyn *hc_prototypes[SCR_DYNH_CAPACITY]; static SCR_TL bool hc_teardown_registered; @@ -73,9 +75,13 @@ static ScrDyn *hc_getter(ScrClosure *closure, ScrDyn *const *args, size_t argc) const ScrDynHandleOps *ops = hc_ops(tag); ScrDyn *self = scr_dyn_this_get(); ScrDyn *result = NULL; - if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { - result = ops->get(self->v.handle.ptr, accessor->name, strlen(accessor->name)); - if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined()); + ScrDyn *receiver = NULL; + if (scr_dyn_handle_class_receiver(tag, self, &receiver)) { + if (receiver) { + result = ops->get(receiver->v.handle.ptr, accessor->name, strlen(accessor->name)); + if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined()); + scr_dyn_release(receiver); + } } else { hc_invalid_this(ops); if (accessor->rejects && ops->rejected) result = ops->rejected(); @@ -245,6 +251,23 @@ void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag) hc_aliases[tag] = (uint16_t)(klass_tag + 1); } +void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle)) { + hc_alias_views[tag] = view; +} + +bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out) { + *out = NULL; + if (self->kind != SCR_DYN_HANDLE) return false; + ScrDynHandleTag tag = self->v.handle.tag; + if (tag == klass_tag) { + *out = scr_dyn_retain(self); + return true; + } + if (hc_aliases[tag] != (uint16_t)(klass_tag + 1) || !hc_alias_views[tag]) return false; + *out = hc_alias_views[tag](self); + return true; +} + ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle) { ScrDynHandleTag tag = handle->v.handle.tag; if (hc_classes[tag]) return tag; diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 00dcc9b41..882d57002 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -164,6 +164,20 @@ struct SfStream { /* Consuming the body takes Uint8Array chunks only: set on a body from a * web ReadableStream and on the tee branches of such a body. */ bool u8_chunks; + /* The web ReadableStream this stream's checked-dynamic handles stand + * for once a member the fetch stream lacks is used (+1, or NULL; see + * sf_stream_web_view). It reads this stream; from then on every member + * and every use of the body go through it. */ + ScrDyn *web_view; + /* A reader of web_view (+1) that keeps it locked while this stream was + * locked when web_view was made: by a reader of its own, released + * later, or by reading the body. */ + ScrDyn *web_view_lock; + /* web_view's reader of this stream, taken at its first pull (owned + * through `reader`). */ + SfReader *view_reader; + /* web_view's source holds a reference to this stream. */ + bool view_source_live; #endif ScrArr *from_array; ScrBytes *from_bytes; @@ -917,6 +931,18 @@ static void sf_stream_drop_source_callbacks( static void sf_stream_release(SfStream *s) { if (!s) return; if (--s->rc > 0) { +#ifdef SCR_HANDLE_CLASSES + /* Only the web view's source still holds the stream: the view no + * longer needs to be found from it, and the view, which reads it, + * keeps it while anything else holds the view. */ + if (s->rc == 1 && s->view_source_live && s->web_view) { + ScrDyn *view = s->web_view, *lock = s->web_view_lock; + s->web_view = s->web_view_lock = NULL; + scr_dyn_release(lock); + scr_dyn_release(view); /* may release s */ + return; + } +#endif sf_response_discard_if_unobserved(s); return; } @@ -933,6 +959,8 @@ static void sf_stream_release(SfStream *s) { scr_dyn_release(s->from_dyn); #ifdef SCR_HANDLE_CLASSES scr_dyn_release(s->web_reader); + scr_dyn_release(s->web_view_lock); + scr_dyn_release(s->web_view); #endif scr_arr_release(s->from_array); scr_bytes_release(s->from_bytes); @@ -1629,6 +1657,97 @@ static ScrDyn *sf_web_body(const ScrDyn *body) { static bool sf_is_web_body(const ScrDyn *body) { return body && body->kind == SCR_DYN_HANDLE && body->v.handle.tag == SCR_DYNH_WS_READABLE; } + +static SfReader *sf_stream_get_reader(SfStream *s); +static bool sf_reader_read_request(SfReader *r, ScrPromise *p, const ScrFetchReadSink *sink, void *sink_ctx); +static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason, bool through_reader); + +/* The web view's source (scr_webstream_readable_over): it reads the stream + * through a reader of its own, taken at the first pull. */ +static bool sf_view_read_into(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx) { + SfStream *s = ctx; + if (!s->view_reader) { + SfReader *r = sf_stream_get_reader(s); + if (!r) return false; + s->view_reader = r; /* the stream owns it as its reader */ + sf_reader_release(r); + } + /* A chunk the request takes at once can close the view, whose source + * then lets the stream go while it still drains. */ + sf_stream_retain(s); + bool ok = sf_reader_read_request(s->view_reader, NULL, sink, sink_ctx); + sf_stream_release(s); + return ok; +} + +static ScrPromise *sf_view_cancel(void *ctx, ScrDyn *reason) { + SfStream *s = ctx; + return sf_stream_cancel(s, reason, s->view_reader != NULL && s->reader == s->view_reader); +} + +static void sf_view_release(void *ctx) { + SfStream *s = ctx; + s->view_source_live = false; + sf_stream_release(s); +} + +static const ScrWebStreamSource sf_view_source = {sf_view_read_into, sf_view_cancel, sf_view_release}; + +/* The web ReadableStream the stream's checked-dynamic handles stand for + * (borrowed), made on first use. Node's Response.body is one; the fetch + * stream answers getReader, cancel and locked itself until a member it + * lacks is used, and a lock it holds then keeps the view locked. */ +static ScrDyn *sf_stream_web_view(SfStream *s) { + if (s->web_view) return s->web_view; + s->web_view = scr_webstream_readable_over(&sf_view_source, sf_stream_retain(s)); + s->view_source_live = true; + if (s->reader || s->internal_lock) { + s->web_view_lock = scr_dyn_invoke(s->web_view, "getReader", NULL, 0, "ReadableStream.getReader"); + if (!s->web_view_lock) return NULL; + } + return s->web_view; +} + +/* Whether `key` names one of the fetch stream's own methods (its ops' + * method table). */ +static bool sf_stream_own_method(const char *key, size_t len) { + ScrDyn probe; + memset(&probe, 0, sizeof probe); + probe.kind = SCR_DYN_HANDLE; + probe.v.handle.tag = SCR_DYNH_WEB_STREAM; + for (const ScrDynHandleMethod *m = scr_dyn_handle_ops_of(&probe)->methods; m && m->name; m++) + if (sf_name(key, len, m->name)) return true; + return false; +} + +/* scr_dyn_handle_class_alias_view for the stream's handles (+1). */ +static ScrDyn *sf_stream_view_of(ScrDyn *handle) { + ScrDyn *view = sf_stream_web_view((SfStream *)handle->v.handle.ptr); + return view ? scr_dyn_retain(view) : NULL; +} + +/* A body whose stream has a web view is read through the view, which + * locks it as reading a body locks Node's stream: a native stream that + * reads the view (+1), or NULL with the TypeError pending. Otherwise the + * stream itself (+1). */ +static SfStream *sf_stream_body_source(SfStream *s) { + if (!s->web_view) return sf_stream_retain(s); + ScrDyn *reading = sf_web_body(s->web_view); + if (!reading) return NULL; + SfStream *out = sf_stream_retain((SfStream *)reading->v.handle.ptr); + scr_dyn_release(reading); + return out; +} +#endif + +/* Whether a body's web view is locked or disturbed, which makes the body + * unusable as one, as its own lock and disturbed state do. */ +#ifdef SCR_HANDLE_CLASSES +static bool sf_view_unusable(const SfStream *s) { + return s->web_view && scr_web_stream_bridge->unusable(s->web_view); +} +#else +#define sf_view_unusable(s) false #endif static void sf_stream_pull_from_source(SfStream *s) { @@ -1882,10 +2001,22 @@ static void sf_reader_release_lock(SfReader *r) { sf_reader_reject_message_all(r, "Invalid state: Releasing reader"); +#ifdef SCR_HANDLE_CLASSES + /* The web view was made while this reader held the lock. */ + ScrDyn *view_lock = r != s->view_reader ? s->web_view_lock : NULL; + if (view_lock) s->web_view_lock = NULL; +#endif size_t handle_refs = r->handle_refs; s->reader = NULL; r->stream = NULL; sf_reader_release(r); /* stream's ownership */ +#ifdef SCR_HANDLE_CLASSES + if (view_lock) { + ScrDyn *released = scr_dyn_invoke(view_lock, "releaseLock", NULL, 0, "ReadableStreamDefaultReader.releaseLock"); + if (released) scr_dyn_release(released); + scr_dyn_release(view_lock); + } +#endif while (handle_refs-- > 0) sf_stream_release(s); } @@ -1999,8 +2130,21 @@ static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason, } static ScrPromise *sf_stream_collect(SfStream *s, int mode) { +#ifdef SCR_HANDLE_CLASSES + if (s->web_view && !(s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s))) { + SfStream *source = sf_stream_body_source(s); + if (!source) { + ScrPromise *p = scr_promise_new(); + scr_promise_reject_pending(p); + return p; + } + ScrPromise *p = sf_stream_collect(source, mode); + sf_stream_release(source); + return p; + } +#endif ScrPromise *p = scr_promise_new(); - if (s->disturbed || s->reader || s->internal_lock) { + if (s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s)) { sf_type_error("Body is unusable: Body has already been read"); scr_promise_reject_pending(p); return p; @@ -2456,6 +2600,14 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { SfStream *s = ptr; +#ifdef SCR_HANDLE_CLASSES + /* Any other member, and every member once the web view exists, is the + * view's. */ + if (s->web_view || !sf_stream_own_method(method, strlen(method))) { + ScrDyn *view = sf_stream_web_view(s); + return view ? scr_dyn_invoke(view, method, args, argc, what) : NULL; + } +#endif if (strcmp(method, "getReader") == 0) { ScrDyn *options = argc ? args[0] : NULL; ScrDyn *materialized = NULL; @@ -2494,12 +2646,35 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method, static ScrDyn *sf_stream_get(void *ptr, const char *key, size_t len) { SfStream *s = ptr; +#ifdef SCR_HANDLE_CLASSES + /* The stream's own members are its own until the web view exists; any + * other member makes the view and is the view's. */ + if (s->web_view || !(sf_name(key, len, "locked") || sf_stream_own_method(key, len))) { + ScrDyn *view = sf_stream_web_view(s); + if (!view) return NULL; + ScrStr *name = scr_str_new(key, len); + ScrDyn *value = scr_dyn_handle_key_get(view, name); + scr_str_release(name); + return value; + } +#endif if (sf_name(key, len, "locked")) { return scr_dyn_new_bool(s->reader != NULL || s->internal_lock); } return NULL; } +#ifdef SCR_HANDLE_CLASSES +/* Symbol.asyncIterator is the web view's. */ +static ScrDyn *sf_stream_symbol_get(void *ptr, const ScrDyn *key) { + if (!scr_dyn_is_async_iterator_symbol(key)) return NULL; + ScrDyn *view = sf_stream_web_view(ptr); + if (!view) return NULL; + const ScrDynHandleOps *ops = scr_dyn_handle_ops_of(view); + return ops->symbol_get ? ops->symbol_get(view->v.handle.ptr, key) : NULL; +} +#endif + static ScrDyn *sf_reader_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { @@ -4098,10 +4273,30 @@ static ScrPromise *sf_response_collect(SfResponse *r, int mode) { * guard. NULL with a TypeError pending when the body is unusable. */ static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) { if (!r->null_body && - (r->body->disturbed || r->body->reader || r->body->internal_lock)) { + (r->body->disturbed || r->body->reader || r->body->internal_lock || sf_view_unusable(r->body))) { sf_type_error(unusable); return NULL; } +#ifdef SCR_HANDLE_CLASSES + /* A body with a web view tees through it; each side reads one branch. */ + SfStream *viewed[2] = {NULL, NULL}; + if (!r->null_body && r->body->web_view) { + ScrDyn *branches = scr_dyn_invoke(r->body->web_view, "tee", NULL, 0, "ReadableStream.tee"); + if (!branches) return NULL; + for (int i = 0; i < 2; i++) { + ScrDyn *reading = sf_web_body(branches->v.arr.items[i]); + if (!reading) { + sf_stream_release(viewed[0]); + scr_dyn_release(branches); + return NULL; + } + viewed[i] = sf_stream_retain((SfStream *)reading->v.handle.ptr); + viewed[i]->u8_chunks = r->body->u8_chunks; + scr_dyn_release(reading); + } + scr_dyn_release(branches); + } +#endif SfResponse *c = calloc(1, sizeof *c); if (!c) sf_oom(); c->rc = 1; @@ -4110,7 +4305,16 @@ static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) { sf_stream_close(c->body); } else { SfStream *branches[2]; +#ifdef SCR_HANDLE_CLASSES + if (viewed[0]) { + branches[0] = viewed[0]; + branches[1] = viewed[1]; + } else { + sf_stream_tee(r->body, branches); + } +#else sf_stream_tee(r->body, branches); +#endif sf_stream_release(r->body); r->body = branches[0]; c->body = branches[1]; @@ -6153,13 +6357,28 @@ static bool sf_start_hop(SfTransfer *t) { stream = sf_stream_of(body); } if (!stream || stream->reader || stream->internal_lock || - stream->disturbed) { + stream->disturbed || sf_view_unusable(stream)) { scr_dyn_release(seeded); if (t->client) scr_http_client_destroy(t->client); sf_reject(t, "Response body object should not be disturbed or locked"); return false; } +#ifdef SCR_HANDLE_CLASSES + /* A stream with a web view uploads through it; the transfer holds the + * stream that reads the view. */ + SfStream *through = NULL; + if (stream->web_view) { + through = sf_stream_body_source(stream); + if (!through) { + scr_dyn_release(seeded); + if (t->client) scr_http_client_destroy(t->client); + sf_reject(t, "Response body object should not be disturbed or locked"); + return false; + } + stream = through; + } +#endif stream->disturbed = true; stream->internal_lock = true; stream->request_owner = t; @@ -6169,6 +6388,9 @@ static bool sf_start_hop(SfTransfer *t) { !stream->error && !t->done && t->client) { sf_stream_pull(stream); } +#ifdef SCR_HANDLE_CLASSES + sf_stream_release(through); +#endif scr_dyn_release(seeded); } else { scr_http_client_end(client); @@ -6429,12 +6651,17 @@ static ScrDyn *sf_response_construct(ScrDyn *body, ScrDyn *init, } else if (body->kind == SCR_DYN_HANDLE && body->v.handle.tag == SCR_DYNH_WEB_STREAM) { stream = (SfStream *)body->v.handle.ptr; - if (stream->reader || stream->internal_lock || stream->disturbed) { + if (stream->reader || stream->internal_lock || stream->disturbed || sf_view_unusable(stream)) { stream = NULL; /* borrowed until the successful retain below */ sf_type_error("Response body object should not be disturbed or locked"); goto fail; } +#ifdef SCR_HANDLE_CLASSES + stream = sf_stream_body_source(stream); + if (!stream) goto fail; +#else sf_stream_retain(stream); +#endif #ifdef SCR_HANDLE_CLASSES } else if (sf_is_web_body(body)) { ScrDyn *web = sf_web_body(body); @@ -6676,7 +6903,8 @@ ScrDyn *scr_fetch_request_new(ScrDyn *input, ScrDyn *init) { ScrDyn *body = sf_response_init_get(init, "body", 4); bool supplied = body && body->kind != SCR_DYN_UNDEF && body->kind != SCR_DYN_NULL; if (!supplied && source && !source->payload->null_body) { - if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock) { + if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock || + sf_view_unusable(source->payload->body)) { scr_dyn_release(body); scr_str_release(method_text); scr_url_release(url); sf_type_error("Cannot construct a Request with a Request object that has already been used."); return NULL; } @@ -7195,6 +7423,9 @@ static const ScrDynHandleOps sf_abort_controller_ops = { static const ScrDynHandleOps sf_stream_ops = { "ReadableStream", &sf_stream_retain_v, &sf_stream_release_v, &sf_stream_invoke, &sf_stream_get, &sf_no_set, NULL, +#ifdef SCR_HANDLE_CLASSES + .symbol_get = &sf_stream_symbol_get, +#endif .methods = sf_stream_methods, .invalid_this = "Value of \"this\" must be of type ReadableStream", .rejected = &sf_rejected}; @@ -7273,6 +7504,9 @@ void scr_fetch_install(void) { scr_dyn_handle_install(SCR_DYNH_ABORT_CONTROLLER, &sf_abort_controller_ops); scr_dyn_handle_install(SCR_DYNH_WEB_STREAM, &sf_stream_ops); +#ifdef SCR_HANDLE_CLASSES + scr_dyn_handle_class_alias_view(SCR_DYNH_WEB_STREAM, &sf_stream_view_of); +#endif scr_dyn_handle_install(SCR_DYNH_WEB_READER, &sf_reader_ops); scr_dyn_handle_install(SCR_DYNH_WEB_CONTROLLER, &sf_controller_ops); scr_dyn_handle_install(SCR_DYNH_FETCH_REQUEST, &sf_request_ops); diff --git a/packages/runtime/src/scr_generator.h b/packages/runtime/src/scr_generator.h index 6b66e8f9d..3a47444e4 100644 --- a/packages/runtime/src/scr_generator.h +++ b/packages/runtime/src/scr_generator.h @@ -233,6 +233,24 @@ typedef struct ScrWebStreamBridge { bool (*unusable)(const ScrDyn *stream); } ScrWebStreamBridge; extern SCR_TL const ScrWebStreamBridge *scr_web_stream_bridge; +/* The read side of a stream another unit implements, which + * scr_webstream_readable_over reads: read_into queues one read request on + * it (false with the exception pending), cancel cancels it (+1 promise), + * and release drops the context. */ +typedef struct ScrWebStreamSource { + bool (*read_into)(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx); + ScrPromise *(*cancel)(void *ctx, ScrDyn *reason); + void (*release)(void *ctx); +} ScrWebStreamSource; +/* A checked-dynamic ReadableStream (+1 handle) with a zero high-water mark + * that reads one chunk of `source` per pull (scr_webstreams.c). ctx's + * reference moves in; the stream releases it when it goes away. */ +ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx); +/* Installs the web stream classes (webstream.install): a program whose + * JavaScript uses a ReadableStream member the fetch unit's streams lack + * calls it at startup, which links the classes that fetch bodies then + * stand for. */ +void scr_webstream_install(void); /* The web stream class constructors for checked-dynamic code, by kind * (scr_webstreams.c: ReadableStream, ReadableStreamDefaultReader, * ReadableStreamDefaultController, CountQueuingStrategy, diff --git a/packages/runtime/src/scr_json.c b/packages/runtime/src/scr_json.c index 1177231ee..76bce14a4 100644 --- a/packages/runtime/src/scr_json.c +++ b/packages/runtime/src/scr_json.c @@ -7015,16 +7015,20 @@ static ScrDyn *scr_dynh_method_call(ScrClosure *closure, ScrDyn *const *args, si const ScrDynHandleOps *ops = scr_dyn_handle_ops(tag); ScrDyn *self = scr_dyn_this_get(); ScrDyn *result; - if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { #ifdef SCR_HANDLE_CLASSES - /* A class's method value reaches the class's method past the handle's - * own properties; this build is the one programs with such classes - * link. */ - result = scr_dyn_handle_class_method_call(ops, self, method->name, args, argc); -#else - result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name); -#endif + /* A class's method value reaches the class's method past the handle's + * own properties, and runs on the instance a handle of an alias tag + * stands for; this build is the one programs with such classes link. */ + ScrDyn *receiver = NULL; + if (scr_dyn_handle_class_receiver(tag, self, &receiver)) { + result = receiver ? scr_dyn_handle_class_method_call(ops, receiver, method->name, args, argc) : NULL; + scr_dyn_release(receiver); } else { +#else + if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) { + result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name); + } else { +#endif const char *message = method->invalid_this ? method->invalid_this : ops->invalid_this ? ops->invalid_this : "Illegal invocation"; diff --git a/packages/runtime/src/scr_webstreams.c b/packages/runtime/src/scr_webstreams.c index 1ea9c754e..43868748e 100644 --- a/packages/runtime/src/scr_webstreams.c +++ b/packages/runtime/src/scr_webstreams.c @@ -1691,6 +1691,96 @@ static ScrDyn *ws_readable_from_thunk(ScrClosure *closure, ScrDyn *const *args, return ws_readable_from(argc > 0 ? args[0] : scr_dyn_undefined()); } +/* ── streams over another unit's reads (scr_webstream_readable_over) ─ */ + +typedef struct WsOver { + WsObj obj; + WsReadable *stream; /* +1 */ + const ScrWebStreamSource *source; + void *ctx; /* released through source->release */ +} WsOver; + +static void ws_over_trace(void *obj, ScrTraceVisit visit, void *ctx) { ws_visit(((WsOver *)obj)->stream, visit, ctx); } + +static void ws_over_drop(void *obj, bool gc) { + WsOver *o = obj; + ws_drop_obj(o->stream, gc); + o->stream = NULL; + /* The context is the other unit's, outside the collector. */ + void *ctx = o->ctx; + o->ctx = NULL; + if (ctx) o->source->release(ctx); +} + +static const WsObjType ws_over_type = {ws_over_trace, ws_over_drop}; + +/* The stream's controller while it still holds this source. */ +static WsRController *ws_over_controller(WsOver *o) { + WsRController *c = o->stream ? o->stream->controller : NULL; + return c && !c->algorithms_cleared ? c : NULL; +} + +static void ws_over_chunk(void *ctx, ScrDyn *value) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) { + if (ws_rcontroller_enqueue(c, value)) { + ws_pull_fulfilled(c, NULL); + } else { + ScrDyn *error = ws_take_pending(); + ws_rcontroller_error(c, error); + scr_dyn_release(error); + } + } + ws_release(o); +} + +static void ws_over_close(void *ctx) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) ws_rcontroller_close(c); + ws_release(o); +} + +static void ws_over_error(void *ctx, ScrDyn *reason) { + WsOver *o = ctx; + WsRController *c = ws_over_controller(o); + if (c) ws_rcontroller_error(c, reason ? reason : scr_dyn_undefined()); + ws_release(o); +} + +static const ScrFetchReadSink ws_over_sink = {ws_over_chunk, ws_over_close, ws_over_error}; + +static ScrDyn *ws_over_pull(void *ctx, WsRController *c) { + (void)c; + WsOver *o = ctx; + if (!o->source->read_into(o->ctx, &ws_over_sink, ws_retain(o))) { + ws_release(o); + return NULL; + } + return WS_PARKED; +} + +static ScrPromise *ws_over_cancel(void *ctx, WsRController *c, ScrDyn *reason) { + (void)c; + WsOver *o = ctx; + return o->source->cancel(o->ctx, reason); +} + +static const WsSourceOps ws_over_source_ops = {ws_over_pull, ws_over_cancel}; + +ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx) { + ws_install(); + WsOver *o = ws_obj_new(sizeof *o, &ws_over_type); + o->source = source; + o->ctx = ctx; + WsReadable *s = ws_readable_create_native(&ws_over_source_ops, o, 0, 0); + o->stream = ws_readable_retain(s); + ScrDyn *box = ws_stream_box(s); + ws_readable_release(s); + return box; +} + /* ── async iteration (ReadableStream.prototype.values) ────────────── */ typedef struct WsIter { @@ -5184,6 +5274,8 @@ static ScrDynHandleTag ws_class_tag(double kind) { return tags[index]; } +void scr_webstream_install(void) { ws_install(); } + /* The web stream class `kind` names as a constructor value (+1): the * compiler's webstream.class. */ ScrDyn *scr_webstream_class(double kind) { diff --git a/tests/corpus/fetch-body-web-stream-linked.mjs b/tests/corpus/fetch-body-web-stream-linked.mjs new file mode 100644 index 000000000..0be52dac9 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-linked.mjs @@ -0,0 +1,93 @@ +// Fetch Response and Request bodies as web ReadableStreams in a program +// that also names the classes: tee, pipeTo, pipeThrough, values and +// for await on a body, the body's lock and bodyUsed across them, and the +// body's other uses (text, clone, a new Response) after one. +const decoder = new TextDecoder(); +const text = (chunks) => chunks.map((c) => decoder.decode(c)).join(""); +const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`); +const caught = (fn) => { + try { + return `ok:${fn()}`; + } catch (e) { + return `${e.name}:${e.message}`; + } +}; +// Response.clone runs on checked-dynamic handles; the clone goes through +// an untyped parameter. +const cloneOf = (value) => value.clone(); +const readAll = async (stream) => { + const reader = stream.getReader(); + const chunks = []; + for (;;) { + const { done, value } = await reader.read(); + if (done) return text(chunks); + chunks.push(value); + } +}; + +{ + const response = new Response("tee body"); + const body = response.body; + console.log("members", typeof body.tee, typeof body.pipeTo, typeof body.pipeThrough, typeof body.values, "tee" in body, body instanceof ReadableStream); + console.log("before", body.locked, response.bodyUsed); + const [a, b] = body.tee(); + console.log("teed", body.locked, response.bodyUsed, a instanceof ReadableStream, response.body === body); + console.log("branches", await readAll(a), await readAll(b), response.bodyUsed); + console.log("used", await settled(response.text()), caught(() => body.tee())); +} + +{ + const chunks = []; + const response = new Response("piped"); + await response.body.pipeTo(new WritableStream({ write(c) { chunks.push(c); } })); + console.log("pipeTo", text(chunks), response.bodyUsed, response.body.locked); +} + +{ + const upper = new TransformStream({ transform(chunk, c) { c.enqueue(decoder.decode(chunk).toUpperCase()); } }); + const out = new Response("through").body.pipeThrough(upper); + const reader = out.getReader(); + console.log("pipeThrough", (await reader.read()).value, (await reader.read()).done); +} + +{ + const request = new Request("http://example.com/", { method: "POST", body: "iterate me", duplex: "half" }); + let joined = ""; + for await (const chunk of request.body) joined += decoder.decode(chunk); + console.log("for await", joined, request.bodyUsed, request.body.locked); + const values = []; + for await (const chunk of new Response("values").body.values()) values.push(decoder.decode(chunk)); + console.log("values", values.join("|")); + const early = new Response("early"); + for await (const chunk of early.body) { + console.log("break", chunk.length); + break; + } + console.log("after break", early.bodyUsed, early.body.locked, await settled(early.text())); +} + +{ + const response = new Response("read first"); + const reader = response.body.getReader(); + console.log("reader first", caught(() => { response.body.tee(); return "teed"; }), response.body.locked); + reader.releaseLock(); + const [a] = response.body.tee(); + console.log("after release", response.body.locked, await readAll(a)); +} + +{ + const response = new Response("member value"); + const tee = response.body.tee; + console.log("method value", tee === ReadableStream.prototype.tee, Array.isArray(tee.call(response.body)), response.body.locked); + const viewed = new Response("viewed"); + void viewed.body.values; + console.log("viewed text", viewed.body.locked, viewed.bodyUsed, await viewed.text(), viewed.bodyUsed, viewed.body.locked); + const cloned = new Response("cloned"); + void cloned.body.tee; + const copy = cloneOf(cloned); + console.log("viewed clone", await copy.text(), await cloned.text()); + const source = new Response("rewrapped"); + void source.body.pipeTo; + const rewrapped = new Response(source.body); + console.log("rewrapped", source.bodyUsed, await rewrapped.text(), source.bodyUsed); +} diff --git a/tests/corpus/fetch-body-web-stream-network.mjs b/tests/corpus/fetch-body-web-stream-network.mjs new file mode 100644 index 000000000..d7e3b59d5 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-network.mjs @@ -0,0 +1,26 @@ +// fetch() response bodies from a local server as web ReadableStreams: for +// await, tee, text() after a member read, and a break out of the +// iteration, with the body's bodyUsed after them. +import http from "node:http"; +const decoder = new TextDecoder(); +const server = http.createServer((req, res) => { + res.write("chunk-one;"); + setTimeout(() => { res.end("chunk-two"); }, 5); +}); +await new Promise((r) => server.listen(0, "127.0.0.1", r)); +const url = `http://127.0.0.1:${server.address().port}/`; +const a = await fetch(url); +let s = ""; +for await (const c of a.body) s += decoder.decode(c); +console.log("for await", s, a.bodyUsed); +const b = await fetch(url); +const [x, y] = b.body.tee(); +const readAll = async (st) => { let t = ""; for await (const c of st) t += decoder.decode(c); return t; }; +console.log("tee", await readAll(x), await readAll(y)); +const c = await fetch(url); +void c.body.values; +console.log("text", await c.text()); +const d = await fetch(url); +for await (const chunk of d.body) { console.log("first", decoder.decode(chunk)); break; } +console.log("after break", d.bodyUsed); +server.close(); diff --git a/tests/corpus/fetch-body-web-stream-unlinked.mjs b/tests/corpus/fetch-body-web-stream-unlinked.mjs new file mode 100644 index 000000000..d9c7f4c38 --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-unlinked.mjs @@ -0,0 +1,25 @@ +// Fetch bodies as web ReadableStreams in a program that names none of the +// web stream classes: tee, values and for await on Response and Request +// bodies, and the body's lock and bodyUsed after them. +const decoder = new TextDecoder(); +const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`); +const drain = async (stream) => { + let out = ""; + for await (const chunk of stream) out += decoder.decode(chunk); + return out; +}; + +const response = new Response("two ways"); +const [left, right] = response.body.tee(); +console.log("tee", await drain(left), await drain(right), response.bodyUsed, response.body.locked); +console.log("used", await settled(response.text())); + +const request = new Request("http://example.com/", { method: "PUT", body: "request body", duplex: "half" }); +console.log("for await", await drain(request.body), request.bodyUsed); + +const parts = []; +for await (const chunk of new Response("values()").body.values({ preventCancel: true })) parts.push(chunk.length); +console.log("values", parts.join(",")); + +const plain = new Response("plain"); +console.log("plain", plain.body.locked, await plain.text(), plain.bodyUsed); diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 } diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 } diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json new file mode 100644 index 000000000..2a2bade0b --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json @@ -0,0 +1 @@ +{ "plain": 3200, "san": 52000 } From 98f5926119011271c07ba1165cd757146583da3e Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Wed, 28 Oct 2026 05:29:42 -0500 Subject: [PATCH 2/2] Update bodyUsed when a body's web view starts a read, and link the view for untyped uses After a fetch body's handle gained its web view, a read, cancel or pipe started on the view left Response.bodyUsed false until the view's pull reached the body, where Node's bodyUsed reads the stream's disturbed state at once. bodyUsed now also answers the view's disturbed state, which the web streams' bridge reports (ScrWebStreamBridge.disturbed). A program that names none of the web stream classes linked them only for member accesses the checker resolves to ReadableStream, so body.tee() or for await over a body inside a function whose parameters have no type threw a TypeError. The same member names on an untyped receiver, and for await over an untyped value, now link the classes in a program that links the fetch unit. Adds corpus program fetch-body-web-stream-helpers.mjs, with its corpus cost and preflight/order baseline, and bodyUsed cases to fetch-body-web-stream-linked.mjs. --- .../lowering/builtins/web-streams-dyn.ts | 57 ++++++++++++++----- .../fetch-body-web-stream-helpers.mjs.json | 6 ++ packages/runtime/src/scr_fetch.c | 6 ++ packages/runtime/src/scr_generator.h | 3 + packages/runtime/src/scr_webstreams.c | 7 ++- .../corpus/fetch-body-web-stream-helpers.mjs | 15 +++++ tests/corpus/fetch-body-web-stream-linked.mjs | 23 ++++++++ .../fetch-body-web-stream-helpers.mjs.json | 1 + 8 files changed, 102 insertions(+), 16 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json create mode 100644 tests/corpus/fetch-body-web-stream-helpers.mjs create mode 100644 tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.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 f18f21e16..765d3beae 100644 --- a/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts +++ b/packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts @@ -1,5 +1,6 @@ import { staticFetchMembers } from "../../../compat/fetch-profile.js"; import { numLit } from "../../../ir/build.js"; +import { everyStmt } from "../../../ir/traverse.js"; import { BOOL, DYN, @@ -253,34 +254,59 @@ function isReadableStreamMember(lowerer: Lowerer, access: ts.PropertyAccessExpre return owner !== undefined && owner.name.text === "ReadableStream"; } +/** Whether a checker type is any or unknown, which a fetch body can be. */ +function isUntypedType(type: ts.Type): boolean { + return (type.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0; +} + +/** Whether the program links the fetch unit (a fetch. entry point), whose + * Response and Request bodies can reach JavaScript as untyped values. */ +function programLinksFetch(functions: IrFunction[]): boolean { + let found = false; + const visitor = { + expr: (e: IrExpr): boolean => { + if (e.kind === "libCall" && e.fn.startsWith("fetch.")) found = true; + return !found; + }, + stmt: (): boolean => !found, + }; + for (const fn of functions) { + for (const s of fn.body) everyStmt(s, visitor); + if (found) return true; + } + return false; +} + /** Whether a JavaScript source of the program uses a ReadableStream member * that the fetch unit's own stream does not implement (the fetch profile's * readableStreamUnsupported): a member access the checker resolves to the - * interface, or `for await` over a ReadableStream value. */ -function programUsesWebStreamOnlyMembers(lowerer: Lowerer): boolean { + * interface, or `for await` over a ReadableStream value. The same member + * names on an untyped receiver, or `for await` over an untyped value, + * count in a program that links the fetch unit, since the value can be a + * fetch body. */ +function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction[]): boolean { const members = staticFetchMembers().readableStreamUnsupported; const iterates = members.has("[Symbol.asyncIterator]"); + let untyped = false; for (const file of lowerer.program.getSourceFiles()) { if (file.isDeclarationFile || !isJsSourceFile(file)) continue; let found = false; ts.walkPreorder(file, (node) => { - if ( - (ts.isPropertyAccessExpression(node) && - members.has(node.name.text) && - isReadableStreamMember(lowerer, node)) || - (iterates && - ts.isForOfStatement(node) && - node.awaitModifier !== undefined && - isReadableStreamType(lowerer, lowerer.typeOf(node.expression))) - ) { - found = true; - return "stop"; + let receiver: ts.Type | null = null; + if (ts.isPropertyAccessExpression(node) && members.has(node.name.text)) { + if (isReadableStreamMember(lowerer, node)) found = true; + else receiver = lowerer.typeOf(node.expression); + } else if (iterates && ts.isForOfStatement(node) && node.awaitModifier !== undefined) { + receiver = lowerer.typeOf(node.expression); + if (isReadableStreamType(lowerer, receiver)) found = true; } + if (found) return "stop"; + if (receiver !== null && isUntypedType(receiver)) untyped = true; return undefined; }); if (found) return true; } - return false; + return untyped && programLinksFetch(functions); } /** Links the web stream classes into a program whose JavaScript uses a @@ -289,7 +315,8 @@ function programUsesWebStreamOnlyMembers(lowerer: Lowerer): boolean { * (scr_fetch.c, sf_stream_web_view). %main installs the classes first. A * program that names one of the classes links them already. */ export function installFetchBodyWebStreams(lowerer: Lowerer, functions: IrFunction[]): void { - if (programNamesWebStreamClass(lowerer) || !programUsesWebStreamOnlyMembers(lowerer)) return; + if (programNamesWebStreamClass(lowerer) || !programUsesWebStreamOnlyMembers(lowerer, functions)) + return; const main = functions.find((fn) => fn.name === ENTRY_NAME); if (!main) return; const loc = main.loc; diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json new file mode 100644 index 000000000..502ef427a --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-web-stream-helpers.mjs" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 882d57002..7dde9c6cb 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -4381,6 +4381,12 @@ static ScrDyn *sf_response_get(void *ptr, const char *key, size_t len) { sf_name(key, len, "statusText")) sf_response_materialize(r); if (sf_name(key, len, "bodyUsed")) { +#ifdef SCR_HANDLE_CLASSES + /* A read, cancel or pipe started on the web view disturbs it at once, + * before its pull reaches this stream. */ + if (!r->null_body && r->body->web_view && scr_web_stream_bridge->disturbed(r->body->web_view)) + return scr_dyn_new_bool(true); +#endif return scr_dyn_new_bool(!r->null_body && r->body->disturbed); } if (sf_name(key, len, "body")) { diff --git a/packages/runtime/src/scr_generator.h b/packages/runtime/src/scr_generator.h index 3a47444e4..bbb75d90d 100644 --- a/packages/runtime/src/scr_generator.h +++ b/packages/runtime/src/scr_generator.h @@ -231,6 +231,9 @@ typedef struct ScrWebStreamBridge { /* Whether a ReadableStream is locked or disturbed, which makes it * unusable as a body. */ bool (*unusable)(const ScrDyn *stream); + /* Whether a ReadableStream is disturbed: a read, a cancel or a pipe + * has started on it. */ + bool (*disturbed)(const ScrDyn *stream); } ScrWebStreamBridge; extern SCR_TL const ScrWebStreamBridge *scr_web_stream_bridge; /* The read side of a stream another unit implements, which diff --git a/packages/runtime/src/scr_webstreams.c b/packages/runtime/src/scr_webstreams.c index 43868748e..f3294faa6 100644 --- a/packages/runtime/src/scr_webstreams.c +++ b/packages/runtime/src/scr_webstreams.c @@ -2137,7 +2137,12 @@ static bool ws_readable_unusable(const ScrDyn *stream) { return !s || ws_readable_locked(s) || s->disturbed; } -static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable}; +static bool ws_readable_disturbed(const ScrDyn *stream) { + WsReadable *s = ws_stream_of(stream); + return s && s->disturbed; +} + +static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable, ws_readable_disturbed}; /* ── WritableStream (lib/internal/webstreams/writablestream.js) ──── */ diff --git a/tests/corpus/fetch-body-web-stream-helpers.mjs b/tests/corpus/fetch-body-web-stream-helpers.mjs new file mode 100644 index 000000000..b9f12a86b --- /dev/null +++ b/tests/corpus/fetch-body-web-stream-helpers.mjs @@ -0,0 +1,15 @@ +// Fetch bodies reached only through untyped helpers, in a program that +// names none of the web stream classes: tee and for await in functions +// whose parameters have no type. +const decoder = new TextDecoder(); +const split = (body) => body.tee(); +const drain = async (stream) => { + let out = ""; + for await (const chunk of stream) out += decoder.decode(chunk); + return out; +}; +const response = new Response("through helpers"); +const [left, right] = split(response.body); +console.log("helpers", await drain(left), await drain(right), response.bodyUsed); +const request = new Request("http://example.com/", { method: "POST", body: "request helper", duplex: "half" }); +console.log("request", await drain(request.body), request.bodyUsed); diff --git a/tests/corpus/fetch-body-web-stream-linked.mjs b/tests/corpus/fetch-body-web-stream-linked.mjs index 0be52dac9..75f3c0952 100644 --- a/tests/corpus/fetch-body-web-stream-linked.mjs +++ b/tests/corpus/fetch-body-web-stream-linked.mjs @@ -91,3 +91,26 @@ const readAll = async (stream) => { const rewrapped = new Response(source.body); console.log("rewrapped", source.bodyUsed, await rewrapped.text(), source.bodyUsed); } + +{ + // Starting a read, a cancel, a pipe or an iteration on the body's stream + // updates bodyUsed when it starts, as in Node. + const read = new Response("read now"); + void read.body.values; + const pending = read.body.getReader().read(); + console.log("bodyUsed read", read.bodyUsed); + await pending; + const canceled = new Response("cancel now"); + void canceled.body.tee; + const cancel = canceled.body.cancel("why"); + console.log("bodyUsed cancel", canceled.bodyUsed); + await cancel; + const piped = new Response("pipe now"); + const pipe = piped.body.pipeTo(new WritableStream()); + console.log("bodyUsed pipe", piped.bodyUsed, piped.body.locked); + await pipe; + const iterated = new Response("iterate now"); + const step = iterated.body.values().next(); + console.log("bodyUsed next", iterated.bodyUsed); + await step; +} diff --git a/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json b/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json new file mode 100644 index 000000000..b65433b72 --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json @@ -0,0 +1 @@ +{ "plain": 2600, "san": 48000 }