From 5372c69459cf4ce27f2d121029b8d451f99a43fe Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Thu, 22 Oct 2026 17:50:29 -0500 Subject: [PATCH] Pass { signal } to stream operator callbacks and Duplex.from bodies --- .../builtin-apis/streams-and-readline.mdx | 7 +- packages/cli/test/native-link-info.test.ts | 4 +- .../ambient/scriptc-node-fallback.d.ts | 4 +- .../compiler/src/backend/llvm/lib-shared.ts | 14 + packages/compiler/src/backend/runtime-abi.ts | 4 +- .../src/backend/runtime-pack-targets.test.ts | 2 +- .../compiler/src/backend/runtime-pack.test.ts | 2 +- .../lowering/class-dynamic-dispatch.ts | 67 +- .../src/frontend/lowering/lower-stream.ts | 331 ++++++-- packages/compiler/src/ir/builtin-effects.ts | 4 + .../compiler/src/ir/builtin-signatures.ts | 43 +- packages/compiler/src/ir/runtime-features.ts | 4 + packages/compiler/src/ir/validate.ts | 27 +- .../test/native-codegen-integration.test.ts | 4 +- .../7433-readable-operator-signal.mjs.json | 6 + .../7434-readable-operator-values.ts.json | 6 + .../7435-readable-flatmap-iterables.mjs.json | 6 + .../7436-readable-read-any-value.mjs.json | 6 + .../corpus/7437-duplex-from-signal.mjs.json | 6 + .../corpus/7438-duplex-from-signal.ts.json | 6 + .../7439-readable-read-string-chunks.ts.json | 6 + .../runtime-darwin-arm64/scripts/build.mjs | 2 +- .../runtime-pack-common/scripts/build.mjs | 2 +- packages/runtime/src/scr_console.c | 2 +- packages/runtime/src/scr_core.h | 2 +- packages/runtime/src/scr_stream.h | 47 +- packages/runtime/src/scr_stream_ops.c | 711 +++++++++++++++--- .../corpus/7433-readable-operator-signal.mjs | 36 + tests/corpus/7434-readable-operator-values.ts | 32 + .../7435-readable-flatmap-iterables.mjs | 29 + tests/corpus/7436-readable-read-any-value.mjs | 17 + tests/corpus/7437-duplex-from-signal.mjs | 54 ++ tests/corpus/7438-duplex-from-signal.ts | 26 + .../7439-readable-read-string-chunks.ts | 17 + .../7433-readable-operator-signal.mjs.json | 1 + .../7434-readable-operator-values.ts.json | 1 + .../7435-readable-flatmap-iterables.mjs.json | 1 + .../7436-readable-read-any-value.mjs.json | 1 + .../7437-duplex-from-signal.mjs.json | 1 + .../7438-duplex-from-signal.ts.json | 1 + .../7439-readable-read-string-chunks.ts.json | 1 + .../self-hosting-native-toolchain.test.ts | 2 +- tests/harness/stream-entry-kinds.test.ts | 77 +- 43 files changed, 1321 insertions(+), 301 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7433-readable-operator-signal.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7434-readable-operator-values.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7435-readable-flatmap-iterables.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7436-readable-read-any-value.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7437-duplex-from-signal.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7438-duplex-from-signal.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7439-readable-read-string-chunks.ts.json create mode 100644 tests/corpus/7433-readable-operator-signal.mjs create mode 100644 tests/corpus/7434-readable-operator-values.ts create mode 100644 tests/corpus/7435-readable-flatmap-iterables.mjs create mode 100644 tests/corpus/7436-readable-read-any-value.mjs create mode 100644 tests/corpus/7437-duplex-from-signal.mjs create mode 100644 tests/corpus/7438-duplex-from-signal.ts create mode 100644 tests/corpus/7439-readable-read-string-chunks.ts create mode 100644 tests/harness/corpus-costs/7433-readable-operator-signal.mjs.json create mode 100644 tests/harness/corpus-costs/7434-readable-operator-values.ts.json create mode 100644 tests/harness/corpus-costs/7435-readable-flatmap-iterables.mjs.json create mode 100644 tests/harness/corpus-costs/7436-readable-read-any-value.mjs.json create mode 100644 tests/harness/corpus-costs/7437-duplex-from-signal.mjs.json create mode 100644 tests/harness/corpus-costs/7438-duplex-from-signal.ts.json create mode 100644 tests/harness/corpus-costs/7439-readable-read-string-chunks.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 aa0afd352..fe6bb2dfb 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -10,7 +10,7 @@ The `objectMode` option, and the `readableObjectMode` and `writableObjectMode` options of a `Duplex`, `Transform`, or `PassThrough`, take `true` or `false` literals in an options object literal; options passed as a value are read when the stream is built. In typed code, a `write` or `transform` callback, or a `_write` or `_transform` method, may declare its chunk with any type that converts to a checked-dynamic value: the chunk converts when the callback runs, and a value of another type throws a TypeError there. An object-mode write passes its encoding argument to `write` and `transform` as Node does (`undefined` when absent, `null` when a callback took its place); a parameter declared `string` receives `""` for `undefined` and `null`, and one declared `string | null | undefined` receives Node's value. - `readable.read()` on an object-mode stream answers its Buffers; another value throws a not-supported error (read such values with `'data'` listeners, `for await`, or `pipe`). + `readable.read()` on an object-mode stream answers any value it holds. `node:stream/consumers` and readline interfaces read object-mode streams of strings and Buffers; other values throw a not-supported error. `pipeline` accepts function stages anywhere among its streams: async generator functions (`async function* (source)`, iterating `source` with `for await`), an async generator source function (`async function* ({ signal })`), an async generator function as the destination, and an async function destination whose value resolves the promise form and reaches the callback's second argument. A stage is written inline, or named by a function declaration or a `const` bound to a function expression that nothing reassigns. @@ -20,10 +20,11 @@ `@types/node` types that result as `NodeJS.WritableStream`, so `for await` over it needs the `PassThrough` type (`pipeline(...) as PassThrough`, or a JSDoc `@type`). In typed code, the callback of a pipeline with an async function destination receives `null` instead of `undefined` when there is no error. `Readable.from` accepts `objectMode` and `highWaterMark` options. - The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them. + The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options. + Their callbacks may take the `{ signal }` argument, `map` may return any value, and `flatMap` may return any iterable, async iterable, or stream. `Duplex.from` accepts an async generator function, which reads what is written to the Duplex, a `Duplex`, `Transform` or `PassThrough`, which it returns, and the values `Readable.from` accepts, which give a Duplex that is not writable. It also accepts a lone `Readable` or `Writable` and a `{ readable, writable }` pair, which the Duplex reads and writes through, a promise, whose value it reads, and an async function, which reads what is written and must resolve to `null` or `undefined`. - A web stream and a body function that declares the options argument are unsupported in `Duplex.from`. + A `Duplex.from` body function may take the `{ signal }` argument, which is aborted with the error the Duplex is destroyed with. `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`. A web stream in a later compose stage, `Readable.toWeb`, and web `WritableStream` and `TransformStream` interop are unsupported. diff --git a/packages/cli/test/native-link-info.test.ts b/packages/cli/test/native-link-info.test.ts index 5b917c63b..70028ab2d 100644 --- a/packages/cli/test/native-link-info.test.ts +++ b/packages/cli/test/native-link-info.test.ts @@ -72,7 +72,7 @@ describe.runIf(supported)("macOS arm64 native link info", () => { minimum_os: "14.0", }, program: { object, entry_symbol: "main" }, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, runtime_pack: { kind: "precompiled", package: "@scriptc/runtime-darwin-arm64", @@ -165,6 +165,6 @@ describe.runIf(supported)("macOS arm64 native link info", () => { () => null, (failure: { stderr?: string }) => failure, ); - expect(error?.stderr).toContain("scr_runtime_abi_v31"); + expect(error?.stderr).toContain("scr_runtime_abi_v32"); }); }); diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index bd8f78105..c2244a233 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/packages/compiler/ambient/scriptc-node-fallback.d.ts @@ -7272,7 +7272,9 @@ declare module "stream" { push(chunk: any, encoding?: string): boolean; unshift(chunk: Buffer | string, encoding?: string): void; unshift(chunk: any, encoding?: string): void; - read(size?: number): Buffer | null; + /* A Buffer, a string once encoded, an object-mode value, or null (as + * @types/node types it). */ + read(size?: number): any; pause(): this; resume(): this; isPaused(): boolean; diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 8e433832a..cacff945d 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -1134,6 +1134,8 @@ export const LIB_FN_SYMS: Record = { "stream.duplexifyPair": "scr_stream_duplexify_pair", "stream.duplexFromPromise": "scr_stream_duplex_from_promise", "stream.duplexFromAsyncFn": "scr_stream_duplex_from_async_fn", + "stream.duplexFromGenSignal": "scr_stream_duplex_from_gen_signal", + "stream.duplexFromAsyncFnSignal": "scr_stream_duplex_from_async_fn_signal", "fetch.readableFromWeb": "scr_stream_readable_from_web", "writable.writeEncoding": "scr_stream_write_encoding", "writable.setDefaultEncoding": "scr_stream_set_default_encoding", @@ -1200,6 +1202,8 @@ export const LIB_FN_SYMS: Record = { "readable.opMap": "scr_stream_op_map", "readable.opFilter": "scr_stream_op_filter", "readable.opFlatMap": "scr_stream_op_flat_map", + "readable.opFlatMapAsyncGen": "scr_stream_op_flat_map_async_gen", + "readable.opFlatMapSyncGen": "scr_stream_op_flat_map_sync_gen", "readable.opDrop": "scr_stream_op_drop", "readable.opTake": "scr_stream_op_take", "readable.opToArray": "scr_stream_op_to_array", @@ -1208,6 +1212,11 @@ export const LIB_FN_SYMS: Record = { "readable.opEvery": "scr_stream_op_every", "readable.opFind": "scr_stream_op_find", "readable.opReduce": "scr_stream_op_reduce", + "readable.opSignalList": "scr_stream_op_signal_list", + "readable.opReduceSignal": "scr_stream_op_reduce_signal", + "readable.opSignalOf": "scr_stream_op_signal_of", + "readable.opOnceOptions": "scr_stream_op_once_options", + "readable.opSignalOpts": "scr_stream_op_signal_opts", "readable.pipe": "scr_stream_pipe", "readable.pipeGeneral": "scr_stream_pipe_general", "writable.cork": "scr_stream_cork", @@ -1711,6 +1720,8 @@ export const USES_TIMERS_LIB_FNS = new Set([ "readable.opMap", "readable.opFilter", "readable.opFlatMap", + "readable.opFlatMapAsyncGen", + "readable.opFlatMapSyncGen", "readable.opDrop", "readable.opTake", "readable.opToArray", @@ -1719,6 +1730,7 @@ export const USES_TIMERS_LIB_FNS = new Set([ "readable.opEvery", "readable.opFind", "readable.opReduce", + "readable.opReduceSignal", "readable.resume", "readable.pipe", "readable.pipeGeneral", @@ -1748,6 +1760,8 @@ export const USES_TIMERS_LIB_FNS = new Set([ "stream.duplexifyPair", "stream.duplexFromPromise", "stream.duplexFromAsyncFn", + "stream.duplexFromGenSignal", + "stream.duplexFromAsyncFnSignal", "fetch.readableFromWeb", "stream.setWritev", "stream.setWritevDefault", diff --git a/packages/compiler/src/backend/runtime-abi.ts b/packages/compiler/src/backend/runtime-abi.ts index 462f15a9e..f84a1c213 100644 --- a/packages/compiler/src/backend/runtime-abi.ts +++ b/packages/compiler/src/backend/runtime-abi.ts @@ -2,8 +2,8 @@ * of the link contract: a program object keeps this symbol undefined and the * matching runtime defines it, so incompatible manual links fail before the * program can start. */ -export const RUNTIME_ABI_VERSION = 31 as const; -export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v31" as const; +export const RUNTIME_ABI_VERSION = 32 as const; +export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v32" as const; /** Object consumption is useful today, but the complete scr_* surface may * still change before 1.0. The versioned marker prevents accidental mixing; diff --git a/packages/compiler/src/backend/runtime-pack-targets.test.ts b/packages/compiler/src/backend/runtime-pack-targets.test.ts index 4292b93b2..f897a2c67 100644 --- a/packages/compiler/src/backend/runtime-pack-targets.test.ts +++ b/packages/compiler/src/backend/runtime-pack-targets.test.ts @@ -52,7 +52,7 @@ async function fixture(target: NativeTargetSpec, workers = false) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple }, macros: { executable: [], excluded: ["SCR_LIB"], sanitizer: "external-toolchain-required" }, flavors: { diff --git a/packages/compiler/src/backend/runtime-pack.test.ts b/packages/compiler/src/backend/runtime-pack.test.ts index a5cb5dabe..903ccfd4d 100644 --- a/packages/compiler/src/backend/runtime-pack.test.ts +++ b/packages/compiler/src/backend/runtime-pack.test.ts @@ -127,7 +127,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, compiler: { command: "clang", identity: "fixture clang", diff --git a/packages/compiler/src/frontend/lowering/class-dynamic-dispatch.ts b/packages/compiler/src/frontend/lowering/class-dynamic-dispatch.ts index af3212ef0..8a4a1e417 100644 --- a/packages/compiler/src/frontend/lowering/class-dynamic-dispatch.ts +++ b/packages/compiler/src/frontend/lowering/class-dynamic-dispatch.ts @@ -85,7 +85,7 @@ import { lowerDynamicEmitterOn, } from "./lower-event-emitter.js"; import { protocolArm, protocolFieldArm } from "./class-protocol-dispatch.js"; -import { STREAM_OPERATORS } from "./lower-stream.js"; +import { STREAM_OPERATORS, operatorSignalOptions, reduceSignalSetup } from "./lower-stream.js"; import { deletedPropertyNames, fixedFieldFamily } from "./js-class-fields.js"; const STREAM_METHODS = new Set([ @@ -5640,13 +5640,16 @@ export class ClassDynamicDispatch { const readableT: IrType = { kind: "object", className: "%Readable" }; const op = (fn: string, args: IrExpr[], type: IrType): IrStmt[] => ret({ kind: "libCall", fn: fn as IrLibFn, args: [receiver, ...args], type, loc }); + // Node passes every callback its { signal }; the callback's + // declared parameters are not known here. + const sig = (): IrExpr => operatorSignalOptions(lowerer, arg(1), loc); switch (method) { case "map": - return op("readable.opMap", [arg(0), arg(1)], readableT); + return op("readable.opMap", [arg(0), arg(1), sig()], readableT); case "filter": - return op("readable.opFilter", [arg(0), arg(1)], readableT); + return op("readable.opFilter", [arg(0), arg(1), sig()], readableT); case "flatMap": - return op("readable.opFlatMap", [arg(0), arg(1)], readableT); + return op("readable.opFlatMap", [arg(0), arg(1), sig()], readableT); case "drop": return op("readable.opDrop", [arg(0), arg(1)], readableT); case "take": @@ -5654,25 +5657,49 @@ export class ClassDynamicDispatch { case "toArray": return op("readable.opToArray", [arg(0)], { kind: "promise", inner: DYN }); case "forEach": - return op("readable.opForEach", [arg(0), arg(1)], { kind: "promise", inner: VOID }); + return op("readable.opForEach", [arg(0), arg(1), sig()], { + kind: "promise", + inner: VOID, + }); case "some": - return op("readable.opSome", [arg(0), arg(1)], { kind: "promise", inner: BOOL }); + return op("readable.opSome", [arg(0), arg(1), sig()], { kind: "promise", inner: BOOL }); case "every": - return op("readable.opEvery", [arg(0), arg(1)], { kind: "promise", inner: BOOL }); + return op("readable.opEvery", [arg(0), arg(1), sig()], { + kind: "promise", + inner: BOOL, + }); case "find": - return op("readable.opFind", [arg(0), arg(1)], { kind: "promise", inner: DYN }); - case "reduce": - return op( - "readable.opReduce", - [ - arg(0), - arg(1), - { kind: "boolLit", value: dispatch.source.args.length >= 2, type: BOOL, loc }, - arg(2), - { kind: "numLit", value: 0, type: F64, loc }, - ], - { kind: "promise", inner: DYN }, - ); + return op("readable.opFind", [arg(0), arg(1), sig()], { kind: "promise", inner: DYN }); + case "reduce": { + // the dispatch function's locals: there is no function + // context while dispatch bodies are built + const setup = reduceSignalSetup(lowerer, arg(2), loc, (name, type) => { + const local = { + id: `${name}.${dispatch.fn.locals.length}`, + name, + type, + mutable: false, + }; + dispatch.fn.locals.push(local); + return local; + }); + return [ + ...setup.stmts, + ...op( + "readable.opReduceSignal", + [ + arg(0), + arg(1), + { kind: "boolLit", value: dispatch.source.args.length >= 2, type: BOOL, loc }, + arg(2), + { kind: "numLit", value: 0, type: F64, loc }, + setup.abort, + setup.cbOpts, + ], + { kind: "promise", inner: DYN }, + ), + ]; + } } } if (method === "destroy" && (!incoming.length || incoming[0]?.kind === "unitLit")) diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 7d15e82bb..b6c6a42dd 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -3923,21 +3923,27 @@ function pipelineBoxHelper( /** `() => ac.abort()` over the pipeline's AbortController: an interned * lifted function takes the controller and answers the closure. */ -function pipelineAbortHelper(lowerer: Lowerer, ac: IrExpr, loc: SrcLoc): IrExpr { - const key = "pipelineAbort"; +function pipelineAbortHelper( + lowerer: Lowerer, + ac: IrExpr, + loc: SrcLoc, + withReason = false, +): IrExpr { + const key = withReason ? "pipelineAbortReason" : "pipelineAbort"; let name = lowerer.valueHelpers.get(key); - const abortT = funcOf([], VOID); + const abortT = funcOf(withReason ? [DYN] : [], VOID); if (!name) { name = keyedHelperName(lowerer, "%stream.pipelineAbort", key); lowerer.valueHelpers.set(key, name); const acLocal: IrLocal = { id: "ac.0", name: "ac", type: DYN, mutable: false, boxed: true }; const acRef = varRef(acLocal.id, DYN, loc); + const reasonLocal: IrLocal = { id: "reason.0", name: "reason", type: DYN, mutable: false }; lowerer.liftedFns.push({ name: `${name}.abort`, - params: [], + params: withReason ? [{ localId: reasonLocal.id, name: reasonLocal.name, type: DYN }] : [], returnType: VOID, captures: [{ localId: acLocal.id, name: acLocal.name, type: DYN }], - locals: [acLocal], + locals: withReason ? [acLocal, reasonLocal] : [acLocal], body: [ { kind: "exprStmt", @@ -3946,7 +3952,7 @@ function pipelineAbortHelper(lowerer: Lowerer, ac: IrExpr, loc: SrcLoc): IrExpr recv: acRef, method: "abort", calleeName: "ac.abort", - args: [], + args: withReason ? [varRef(reasonLocal.id, DYN, loc)] : [], type: DYN, loc, }, @@ -4165,11 +4171,11 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE "write the body as async function* (source) { ... } or async function (source) { ... }", ); } - if (fn.parameters.nodes.length > 1) { + if (fn.parameters.nodes.length > 2) { lowerer.noLowering( - "a Duplex.from body that declares the options argument", - fn.parameters.nodes[1]!, - "the body takes only its source", + "a Duplex.from body with parameters after the options argument", + fn.parameters.nodes[2]!, + "the body takes the source and { signal }", ); } const sourceT: SyncGenType = { @@ -4180,6 +4186,46 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE nextT: UNDEFINED_T, }; const lowered = lowerStageFunction(lowerer, fn, sourceT); + const params = lowered.closure.type.kind === "func" ? lowered.closure.type.params : []; + // fromAsyncGen calls the body with { signal } of an AbortController + // that destroy aborts with its error. + let signalArgs: IrExpr[] = []; + let signalDecls: IrStmt[] = []; + if (params.length > 1) { + if (params[1]!.kind !== "dyn") { + lowerer.noLowering( + "a Duplex.from body's options argument with this type", + fn.parameters.nodes[1]!, + "type it as { signal: AbortSignal }", + ); + } + const acLocal = lowerer.declareHiddenLocal("%duplexAbort", DYN); + const ac = varRef(acLocal.id, DYN, loc); + signalDecls = [ + { + kind: "varDecl", + localId: acLocal.id, + init: { kind: "libCall", fn: "fetch.abortControllerNew", args: [], type: DYN, loc }, + loc, + }, + ]; + signalArgs = [ + pipelineAbortHelper(lowerer, ac, loc, true), + { kind: "dynKeyGet", value: ac, key: strLit("signal", loc), type: DYN, loc }, + ]; + } + const withSignal = (fnName: IrLibFn, args: IrExpr[]): IrExpr => { + const call: IrExpr = { + kind: "libCall", + fn: signalArgs.length > 0 ? (`${fnName}Signal` as IrLibFn) : fnName, + args: [...args, ...signalArgs], + type: duplex, + loc, + }; + return signalDecls.length > 0 + ? { kind: "seqExpr", stmts: signalDecls, result: call, type: duplex, loc } + : call; + }; if (!isGen) { // an async function: a writable Duplexify whose final waits for // the function's promise @@ -4200,26 +4246,18 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE node, ); } - return { - kind: "libCall", - fn: "stream.duplexFromAsyncFn", - args: [lowered.closure, asyncSource, pipelineBoxHelper(lowerer, out, loc)], - type: duplex, - loc, - }; + return withSignal("stream.duplexFromAsyncFn", [ + lowered.closure, + asyncSource, + pipelineBoxHelper(lowerer, out, loc), + ]); } const helpers = lowered.out.kind === "generator" ? fromGenHelpers(lowerer, lowered.out, loc) : null; const source = duplexSourceHelper(lowerer, sourceT, loc); if (!helpers || !source) lowerer.noLowering("Duplex.from over a function with this generator type", node); - return { - kind: "libCall", - fn: "stream.duplexFromGen", - args: [lowered.closure, source!, ...helpers!], - type: duplex, - loc, - }; + return withSignal("stream.duplexFromGen", [lowered.closure, source!, ...helpers!]); } // duplexify.js: a stream answers itself (a Duplex) or is wrapped; any // other body reads as from(Duplexify, body, { objectMode: true, @@ -4510,39 +4548,38 @@ export const STREAM_OPERATORS: ReadonlySet = new Set([ "reduce", ]); -/** An operator callback as a checked-dynamic function value. Callbacks - * that declare the options argument ({ signal }) are refused: the - * AbortSignal Node passes there has no representation. map and flatMap - * callbacks must produce strings or Buffers (flatMap: arrays of them, or - * strings) — or checked-dynamic values, which the runtime checks. */ +/** An operator callback as a checked-dynamic function value, and whether + * it declares the options argument ({ signal }), which the runtime then + * passes. Its results convert to checked-dynamic values (flatMap's are + * iterated with yield*). */ function lowerOperatorCallback( lowerer: Lowerer, node: ts.Expression, member: string, maxParams: number, -): IrExpr { +): { fn: IrExpr; takesOptions: boolean; gen?: IrType & { kind: "generator" } } { const loc = locOf(node); const optionsParam = (at: ts.Node): never => lowerer.noLowering( - `${member} callbacks that declare the options argument`, + `${member} callbacks with parameters after the options argument`, at, - "the { signal } argument has no lowering — take only the chunk", + "the callback takes the chunk and { signal }", ); - if ( - (ts.isArrowFunction(node) || ts.isFunctionExpression(node)) && - node.parameters.nodes.length > maxParams - ) { - optionsParam(node.parameters.nodes[maxParams]!); + let takesOptions = false; + if (ts.isArrowFunction(node) || ts.isFunctionExpression(node)) { + if (node.parameters.nodes.length > maxParams + 1) + optionsParam(node.parameters.nodes[maxParams + 1]!); + takesOptions = node.parameters.nodes.length > maxParams; } const v = lowerer.lowerExpr(node); - if (v.type.kind === "dyn") return v; + if (v.type.kind === "dyn") return { fn: v, takesOptions }; if ( v.type.kind !== "func" && isJsSourceFile(ts.getSourceFileOfNode(node)) && lowerer.dynConvertible(v.type) ) { // JavaScript passes any value; Node's validateFunction throws. - return { kind: "dynFrom", value: v, type: DYN, loc }; + return { fn: { kind: "dynFrom", value: v, type: DYN, loc }, takesOptions: false }; } if (v.type.kind !== "func") { lowerer.noLowering( @@ -4551,25 +4588,8 @@ function lowerOperatorCallback( "pass a function", ); } - if (v.type.params.length > maxParams) optionsParam(node); - if (member === "map" || member === "flatMap") { - const ret = v.type.ret.kind === "promise" ? v.type.ret.inner : v.type.ret; - const chunk = (t: IrType): boolean => - t.kind === "string" || t.kind === "dyn" || (t.kind === "bytes" && t.elem === "u8"); - const ok = - member === "map" - ? chunk(ret) - : ret.kind === "dyn" || ret.kind === "string" || (ret.kind === "array" && chunk(ret.elem)); - if (!ok) { - lowerer.noLowering( - `${member} callbacks that return '${lowerer.fmt(ret)}'`, - node, - member === "map" - ? "map results are stream chunks: strings or Buffers (object-mode streams of other values are not lowered)" - : "flatMap results are arrays of strings or Buffers, or strings", - ); - } - } + if (v.type.params.length > maxParams + 1) optionsParam(node); + if (v.type.params.length > maxParams) takesOptions = true; const getRecord = (id: string) => lowerer.shapes.get(id); const getUnion = (id: string) => lowerer.unions.get(id); if (!canBoxFuncIntoDyn(v.type, getRecord, getUnion)) { @@ -4579,7 +4599,95 @@ function lowerOperatorCallback( "the callback's parameter and result types must cross the checked-dynamic boundary", ); } - return { kind: "dynFrom", value: v, type: DYN, loc }; + // flatMap over a compiled generator: the runtime steps it through the + // helpers over its type + const gen = member === "flatMap" && v.type.ret.kind === "generator" ? v.type.ret : undefined; + return { fn: { kind: "dynFrom", value: v, type: DYN, loc }, takesOptions, ...(gen && { gen }) }; +} + +/** The operator callbacks' `{ signal }` (operators.js): map and the + * operators over it pass `{ signal: AbortSignal.any([options?.signal] + * .filter(Boolean)) }`. options is the operator's options value. */ +export function operatorSignalOptions(lowerer: Lowerer, options: IrExpr, loc: SrcLoc): IrExpr { + void lowerer; + const list: IrExpr = { + kind: "libCall", + fn: "readable.opSignalList", + args: [options], + type: DYN, + loc, + }; + const signal: IrExpr = { kind: "libCall", fn: "fetch.abortAny", args: [list], type: DYN, loc }; + return { kind: "libCall", fn: "readable.opSignalOpts", args: [signal], type: DYN, loc }; +} + +/** reduce's signal for a reducer that reads it (operators.js): `const ac + * = new AbortController()`, `options.signal.addEventListener('abort', () + * => ac.abort(), { once: true })`, then reduceSignal's ac.abort() closure + * and `{ signal: ac.signal }`. options is the bound options value. */ +export function reduceSignalSetup( + lowerer: Lowerer, + options: IrExpr, + loc: SrcLoc, + declare: (name: string, type: IrType) => IrLocal = (name, type) => + lowerer.declareHiddenLocal(name, type), +): { stmts: IrStmt[]; abort: IrExpr; cbOpts: IrExpr } { + const acLocal = declare("%opAbort", DYN); + const ac = varRef(acLocal.id, DYN, loc); + const sigLocal = declare("%opUserSignal", DYN); + const sig = varRef(sigLocal.id, DYN, loc); + const abort = pipelineAbortHelper(lowerer, ac, loc); + const stmts: IrStmt[] = [ + { + kind: "varDecl", + localId: acLocal.id, + init: { kind: "libCall", fn: "fetch.abortControllerNew", args: [], type: DYN, loc }, + loc, + }, + { + kind: "varDecl", + localId: sigLocal.id, + init: { kind: "libCall", fn: "readable.opSignalOf", args: [options], type: DYN, loc }, + loc, + }, + { + kind: "if", + cond: { kind: "dynTest", test: "truthy", value: sig, type: BOOL, loc }, + then: [ + { + kind: "exprStmt", + expr: { + kind: "dynInvoke", + recv: sig, + method: "addEventListener", + calleeName: "options.signal.addEventListener", + args: [ + { kind: "dynFrom", value: strLit("abort", loc), type: DYN, loc }, + { kind: "dynFrom", value: pipelineAbortHelper(lowerer, ac, loc), type: DYN, loc }, + { kind: "libCall", fn: "readable.opOnceOptions", args: [], type: DYN, loc }, + ], + type: DYN, + loc, + }, + loc, + }, + ], + else_: null, + loc, + }, + ]; + const signal: IrExpr = { + kind: "dynKeyGet", + value: ac, + key: strLit("signal", loc), + type: DYN, + loc, + }; + return { + stmts, + abort, + cbOpts: { kind: "libCall", fn: "readable.opSignalOpts", args: [signal], type: DYN, loc }, + }; } /** An operator's options (or reduce's initial value): an object literal @@ -4618,21 +4726,37 @@ function lowerStreamOperator( ): IrExpr { const loc = locOf(call); const args = call.arguments.nodes; - const recv = streamReceiverExpr(lowerer, access.expression); + // JS order: the receiver, the callback, then the options; the + // callback's { signal } is built over the options value afterwards. + const decls: IrStmt[] = []; + const bind = (value: IrExpr, name: string): IrExpr => { + const local = lowerer.declareHiddenLocal(name, value.type); + decls.push({ kind: "varDecl", localId: local.id, init: value, loc }); + return varRef(local.id, value.type, loc); + }; + const recv = bind(streamReceiverExpr(lowerer, access.expression), "%opRecv"); const readable: IrType = { kind: "object", className: "%Readable" }; const opts = (i: number): IrExpr => lowerOperatorValue(lowerer, args[i], `${member} options`, loc, true); + let takesOptions = false; + let callbackGen: (IrType & { kind: "generator" }) | undefined; const fnArg = (maxParams: number): IrExpr => { if (args[0] === undefined) return dynUndefinedExpr(loc); - return lowerOperatorCallback(lowerer, args[0], member, maxParams); + const lowered = lowerOperatorCallback(lowerer, args[0], member, maxParams); + takesOptions = lowered.takesOptions; + callbackGen = lowered.gen; + return bind(lowered.fn, "%opFn"); + }; + const lib = (fn: string, a: IrExpr[], type: IrType): IrExpr => { + const call: IrExpr = { kind: "libCall", fn: fn as IrLibFn, args: a, type, loc }; + return { kind: "seqExpr", stmts: decls, result: call, type, loc }; + }; + // the options argument (bound), and the callback's { signal } when it + // declares one + const withSignal = (optsValue: IrExpr): IrExpr[] => { + const o = bind(optsValue, "%opOptions"); + return [o, takesOptions ? operatorSignalOptions(lowerer, o, loc) : dynUndefinedExpr(loc)]; }; - const lib = (fn: string, a: IrExpr[], type: IrType): IrExpr => ({ - kind: "libCall", - fn: fn as IrLibFn, - args: a, - type, - loc, - }); switch (member) { case "map": case "filter": @@ -4644,7 +4768,16 @@ function lowerStreamOperator( : member === "filter" ? "readable.opFilter" : "readable.opFlatMap"; - return lib(fnName, [recv, fn, opts(1)], readable); + const rest = withSignal(opts(1)); + const genHelpers = callbackGen ? fromGenHelpers(lowerer, callbackGen, loc) : null; + if (callbackGen && genHelpers) { + return lib( + callbackGen.async ? "readable.opFlatMapAsyncGen" : "readable.opFlatMapSyncGen", + [recv, fn, ...rest, ...genHelpers], + readable, + ); + } + return lib(fnName, [recv, fn, ...rest], readable); } case "drop": case "take": { @@ -4657,15 +4790,24 @@ function lowerStreamOperator( } case "toArray": return lib("readable.opToArray", [recv, opts(0)], { kind: "promise", inner: DYN }); - case "forEach": - return lib("readable.opForEach", [recv, fnArg(1), opts(1)], { kind: "promise", inner: VOID }); + case "forEach": { + const fn = fnArg(1); + return lib("readable.opForEach", [recv, fn, ...withSignal(opts(1))], { + kind: "promise", + inner: VOID, + }); + } case "some": case "every": { const fn = fnArg(1); - return lib(member === "some" ? "readable.opSome" : "readable.opEvery", [recv, fn, opts(1)], { - kind: "promise", - inner: BOOL, - }); + return lib( + member === "some" ? "readable.opSome" : "readable.opEvery", + [recv, fn, ...withSignal(opts(1))], + { + kind: "promise", + inner: BOOL, + }, + ); } case "find": { const t = lowerer.typeOf(call); @@ -4677,7 +4819,11 @@ function lowerStreamOperator( "the result is a checked-dynamic value — use a boolean predicate and narrow the result", ); } - return lib("readable.opFind", [recv, fnArg(1), opts(1)], { kind: "promise", inner: DYN }); + const fn = fnArg(1); + return lib("readable.opFind", [recv, fn, ...withSignal(opts(1))], { + kind: "promise", + inner: DYN, + }); } case "reduce": { const t = lowerer.typeOf(call); @@ -4704,12 +4850,19 @@ function lowerStreamOperator( } } const fn = fnArg(2); - const init = lowerOperatorValue(lowerer, args[1], "reduce's initial value", loc, false); - return lib( - "readable.opReduce", - [recv, fn, init, boolLit(args.length >= 2, loc), opts(2), numLit(kind, loc)], - { kind: "promise", inner }, + const init = bind( + lowerOperatorValue(lowerer, args[1], "reduce's initial value", loc, false), + "%opInit", ); + const o = bind(opts(2), "%opOptions"); + const head = [recv, fn, init, boolLit(args.length >= 2, loc), o, numLit(kind, loc)]; + if (!takesOptions) return lib("readable.opReduce", head, { kind: "promise", inner }); + const setup = reduceSignalSetup(lowerer, o, loc); + decls.push(...setup.stmts); + return lib("readable.opReduceSignal", [...head, setup.abort, setup.cbOpts], { + kind: "promise", + inner, + }); } } throw new InternalCompilerError(`unknown stream operator ${member}`); @@ -5253,6 +5406,22 @@ export function lowerStreamMethodCall( requireSide(readableSide, "read"); if (args.length > 1) lowerer.noLowering(`read with ${args.length} arguments`, call); const receiver = streamReceiverExpr(lowerer, access.expression); + // JavaScript (and a result the declarations type as any) reads any + // chunk: a Buffer, a string once encoded, or an object-mode value. + const resultTs = lowerer.typeOf(call); + if ( + isJsSourceFile(ts.getSourceFileOfNode(call)) || + (resultTs.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0 + ) { + const sizeDyn = args[0] ? lowerer.lowerExprExpecting(args[0], DYN) : dynUndefinedExpr(loc); + return { + kind: "libCall", + fn: "readable.readDyn", + args: [receiver, sizeDyn], + type: DYN, + loc, + }; + } const size = args[0] ? lowerer.lowerExprExpecting(args[0], F64) : numLit(-1, loc); const type: IrType = unionOf(lowerer, [BYTES, { kind: "nullT" }]); const read: IrExpr = { diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index af2f2a7cc..3e145f485 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -386,6 +386,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "stream.duplexFromDyn", "stream.duplexFromPromise", "stream.duplexFromAsyncFn", + "stream.duplexFromGenSignal", + "stream.duplexFromAsyncFnSignal", "fetch.readableFromWeb", "stream.destroy", "stream.destroyErr", @@ -521,6 +523,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "readable.opMap", "readable.opFilter", "readable.opFlatMap", + "readable.opFlatMapAsyncGen", + "readable.opFlatMapSyncGen", "readable.opDrop", "readable.opTake", "rl.createSignal", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 9b23a40da..68c3eec77 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -4082,6 +4082,13 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ /** Duplex.from(asyncFunction): the stage, its source helper and the * promise's box helper. */ "stream.duplexFromAsyncFn": { argTypes: [null, null, null], result: VOID }, + /** The same for a body that declares the options argument: then the + * AbortController's abort() and its signal. */ + "stream.duplexFromGenSignal": { + argTypes: [null, null, null, null, null, null, null, DYN], + result: VOID, + }, + "stream.duplexFromAsyncFnSignal": { argTypes: [null, null, null, null, DYN], result: VOID }, "writable.writeEncoding": { argTypes: [null, STRING], result: VOID }, "writable.setDefaultEncoding": { argTypes: [null, DYN], result: VOID }, /** Whether a readable.nextChunkDyn step answered the end (the done @@ -4119,17 +4126,39 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ * forEach nothing; reduce: [stream, fn, init, hasInit, options, kind] * with the payload kind (0 dyn, 1 number, 2 boolean, 3 string). The * promise-returning ones reject instead of throwing. */ - "readable.opMap": { argTypes: [null, DYN, DYN], result: VOID }, - "readable.opFilter": { argTypes: [null, DYN, DYN], result: VOID }, - "readable.opFlatMap": { argTypes: [null, DYN, DYN], result: VOID }, + "readable.opMap": { argTypes: [null, DYN, DYN, DYN], result: VOID }, + "readable.opFilter": { argTypes: [null, DYN, DYN, DYN], result: VOID }, + "readable.opFlatMap": { argTypes: [null, DYN, DYN, DYN], result: VOID }, + /** flatMap whose callback returns a compiled generator: the helpers + * over its type follow (async: next, read, return, throw; sync: next, + * return, throw). */ + "readable.opFlatMapAsyncGen": { + argTypes: [null, DYN, DYN, DYN, null, null, null, null], + result: VOID, + }, + "readable.opFlatMapSyncGen": { argTypes: [null, DYN, DYN, DYN, null, null, null], result: VOID }, "readable.opDrop": { argTypes: [null, DYN, DYN], result: VOID }, "readable.opTake": { argTypes: [null, DYN, DYN], result: VOID }, "readable.opToArray": { argTypes: [null, DYN], result: { kind: "promise", inner: DYN } }, - "readable.opForEach": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: VOID } }, - "readable.opSome": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } }, - "readable.opEvery": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } }, - "readable.opFind": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: DYN } }, + "readable.opForEach": { + argTypes: [null, DYN, DYN, DYN], + result: { kind: "promise", inner: VOID }, + }, + "readable.opSome": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: BOOL } }, + "readable.opEvery": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: BOOL } }, + "readable.opFind": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: DYN } }, + /** The map family's last argument is the callback's { signal } (or + * undefined). reduceSignal adds ac.abort() (a closure) and + * { signal: ac.signal } for a reducer that reads them. */ "readable.opReduce": { argTypes: [null, DYN, DYN, BOOL, DYN, F64], result: VOID }, + "readable.opReduceSignal": { + argTypes: [null, DYN, DYN, BOOL, DYN, F64, null, DYN], + result: VOID, + }, + "readable.opSignalList": { argTypes: [DYN], result: DYN }, + "readable.opSignalOpts": { argTypes: [DYN], result: DYN }, + "readable.opSignalOf": { argTypes: [DYN], result: DYN }, + "readable.opOnceOptions": { argTypes: [], result: DYN }, "readable.resume": { argTypes: [null], result: VOID }, "readable.isPaused": { argTypes: [null], result: BOOL }, "readable.pipe": { argTypes: [null, null, BOOL], result: VOID }, diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 12baeea57..68020a1ce 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -223,6 +223,8 @@ const STREAM_LIB_FNS: ReadonlySet = new Set([ "stream.duplexFromDyn", "stream.duplexFromPromise", "stream.duplexFromAsyncFn", + "stream.duplexFromGenSignal", + "stream.duplexFromAsyncFnSignal", "fetch.readableFromWeb", "stream.duplexifyReadable", "stream.duplexifyWritable", @@ -285,6 +287,8 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet = new Set([ "stream.duplexFromDyn", "stream.duplexFromPromise", "stream.duplexFromAsyncFn", + "stream.duplexFromGenSignal", + "stream.duplexFromAsyncFnSignal", ]); /** libCall name prefixes of the Readable entry points that await through the dyn-async unit. */ diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 8c9a0432e..6e8c50ff1 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5520,6 +5520,8 @@ function validateFunction( e.fn === "stream.duplexifyPair" || e.fn === "stream.duplexFromPromise" || e.fn === "stream.duplexFromAsyncFn" || + e.fn === "stream.duplexFromGenSignal" || + e.fn === "stream.duplexFromAsyncFnSignal" || e.fn === "stream.setConstruct" || e.fn === "stream.setWritev" || e.fn === "stream.setWritevDefault" || @@ -5597,7 +5599,9 @@ function validateFunction( if ( e.fn === "stream.duplexFromGen" || e.fn === "stream.duplexFromDyn" || - e.fn === "stream.duplexFromAsyncFn" + e.fn === "stream.duplexFromAsyncFn" || + e.fn === "stream.duplexFromGenSignal" || + e.fn === "stream.duplexFromAsyncFnSignal" ) { // Duplex.from: (stage, source, helpers...) or (body) → a Duplex. if (!isStreamObject(e.type)) { @@ -5786,6 +5790,8 @@ function validateFunction( e.fn === "readable.opMap" || e.fn === "readable.opFilter" || e.fn === "readable.opFlatMap" || + e.fn === "readable.opFlatMapAsyncGen" || + e.fn === "readable.opFlatMapSyncGen" || e.fn === "readable.opDrop" || e.fn === "readable.opTake" ) { @@ -5794,13 +5800,30 @@ function validateFunction( } return true; } - if (e.fn === "readable.opReduce") { + if (e.fn === "readable.opReduce" || e.fn === "readable.opReduceSignal") { const inner = e.type.kind === "promise" ? e.type.inner : null; const kind = e.args[5]?.kind === "numLit" ? e.args[5].value : -1; const want = [DYN, F64, BOOL, STRING][kind]; if (!isStreamObject(e.args[0]?.type) || !inner || !want || !typeEquals(inner, want)) { err(`libCall readable.opReduce must return the promise its kind names`, e.loc); } + if (e.fn === "readable.opReduceSignal" && e.args[6]?.type.kind !== "func") + err(`libCall ${e.fn} takes ac.abort() as a closure`, e.loc); + return true; + } + if (e.fn === "readable.opOnceOptions") { + if (e.args.length !== 0 || e.type.kind !== "dyn") + err(`libCall ${e.fn} takes no arguments`, e.loc); + return true; + } + if ( + e.fn === "readable.opSignalList" || + e.fn === "readable.opSignalOpts" || + e.fn === "readable.opSignalOf" + ) { + // The operator callbacks' { signal }: (dyn) → dyn. + if (e.args.length !== 1 || e.args[0]?.type.kind !== "dyn" || e.type.kind !== "dyn") + err(`libCall ${e.fn} takes and returns a checked-dynamic value`, e.loc); return true; } if (e.fn === "readable.fromDone") { diff --git a/packages/compiler/test/native-codegen-integration.test.ts b/packages/compiler/test/native-codegen-integration.test.ts index 7f819bdad..720de5d6d 100644 --- a/packages/compiler/test/native-codegen-integration.test.ts +++ b/packages/compiler/test/native-codegen-integration.test.ts @@ -378,7 +378,7 @@ done: expect(await symbols(helperObject, ["-u"])).toEqual(await symbols(clangObject, ["-u"])); expect(await symbols(helperObject, ["-gU"])).toEqual(await symbols(clangObject, ["-gU"])); expect(await symbols(helperObject, ["-gU"])).toEqual(["0000000000000000 T _main"]); - expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v31"); + expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v32"); const clangExe = join(dir, "clang-program"); const linkDriver = join(dir, "clang-driver.c"); @@ -712,7 +712,7 @@ done: .split("\n"); expect(undefinedSymbols).toContain("_sf_scale"); expect(undefinedSymbols).toContain("_sf_callback_mix"); - expect(undefinedSymbols).toContain("_scr_runtime_abi_v31"); + expect(undefinedSymbols).toContain("_scr_runtime_abi_v32"); }); test("helper and clang object paths retain the same outbound FFI C ABI", async () => { diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7433-readable-operator-signal.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7433-readable-operator-signal.mjs.json new file mode 100644 index 000000000..5de5d9839 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7433-readable-operator-signal.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7433-readable-operator-signal.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7434-readable-operator-values.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7434-readable-operator-values.ts.json new file mode 100644 index 000000000..c374d9f03 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7434-readable-operator-values.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7434-readable-operator-values.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7435-readable-flatmap-iterables.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7435-readable-flatmap-iterables.mjs.json new file mode 100644 index 000000000..0c3697807 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7435-readable-flatmap-iterables.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7435-readable-flatmap-iterables.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7436-readable-read-any-value.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7436-readable-read-any-value.mjs.json new file mode 100644 index 000000000..6c2f63e17 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7436-readable-read-any-value.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7436-readable-read-any-value.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7437-duplex-from-signal.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7437-duplex-from-signal.mjs.json new file mode 100644 index 000000000..3b0699407 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7437-duplex-from-signal.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7437-duplex-from-signal.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7438-duplex-from-signal.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7438-duplex-from-signal.ts.json new file mode 100644 index 000000000..0702ed0da --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7438-duplex-from-signal.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7438-duplex-from-signal.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7439-readable-read-string-chunks.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7439-readable-read-string-chunks.ts.json new file mode 100644 index 000000000..e3dab7078 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7439-readable-read-string-chunks.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7439-readable-read-string-chunks.ts" + ], + "diags": [] +} diff --git a/packages/runtime-darwin-arm64/scripts/build.mjs b/packages/runtime-darwin-arm64/scripts/build.mjs index e04a0dfcf..760a60553 100644 --- a/packages/runtime-darwin-arm64/scripts/build.mjs +++ b/packages/runtime-darwin-arm64/scripts/build.mjs @@ -340,7 +340,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: RUNTIME_PACK_MATRIX.target, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime-pack-common/scripts/build.mjs b/packages/runtime-pack-common/scripts/build.mjs index dd34cf10b..dc4dc74ef 100644 --- a/packages/runtime-pack-common/scripts/build.mjs +++ b/packages/runtime-pack-common/scripts/build.mjs @@ -437,7 +437,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: matrix.target, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime/src/scr_console.c b/packages/runtime/src/scr_console.c index 194a14b26..e24949ad3 100644 --- a/packages/runtime/src/scr_console.c +++ b/packages/runtime/src/scr_console.c @@ -148,7 +148,7 @@ static void scr_enable_utf8_console_output(void) { } #endif -void scr_runtime_abi_v31(void) {} +void scr_runtime_abi_v32(void) {} void scr_init(void) { #ifdef SCR_WORKERS diff --git a/packages/runtime/src/scr_core.h b/packages/runtime/src/scr_core.h index 9f27a15b1..3d11868a2 100644 --- a/packages/runtime/src/scr_core.h +++ b/packages/runtime/src/scr_core.h @@ -81,7 +81,7 @@ void scr_init(void); /* Program objects emitted by the bundled LLVM helper reference this symbol. * Its versioned spelling makes a mismatched manual runtime link fail before * the program can start. */ -void scr_runtime_abi_v31(void); +void scr_runtime_abi_v32(void); void scr_runtime_workers_v8(void); /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ────── diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index cdc51466e..d26077b5c 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -431,18 +431,41 @@ int scr_stream_push_object(ScrStream *s, const ScrDyn *v); * dyn array, find and reduce a dyn value — reduce's `kind` picks the * payload instead: 0 dyn, 1 number, 2 boolean, 3 string — some and every * a boolean, forEach nothing. */ -ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options); -ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options); -ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options); +ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +/* flatMap whose callback returns a compiled generator: the helpers over its + * type (async: next, read, return, throw; sync: next, return, throw; + * borrowed). */ +ScrStream *scr_stream_op_flat_map_async_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options, + ScrDyn *cb_opts, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, ScrClosure *thr); +ScrStream *scr_stream_op_flat_map_sync_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options, + ScrDyn *cb_opts, ScrClosure *next, ScrClosure *ret, + ScrClosure *thr); ScrStream *scr_stream_op_drop(ScrStream *s, ScrDyn *number, ScrDyn *options); ScrStream *scr_stream_op_take(ScrStream *s, ScrDyn *number, ScrDyn *options); ScrPromise *scr_stream_op_to_array(ScrStream *s, ScrDyn *options); -ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options); -ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options); -ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options); -ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options); +ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); +ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts); ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, ScrDyn *options, double kind); +/* reduce whose reducer reads its { signal } (cb_opts): abort is ac.abort() + * (borrowed), run when the reduction ends. */ +ScrPromise *scr_stream_op_reduce_signal(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, + ScrDyn *options, double kind, ScrClosure *abort, + ScrDyn *cb_opts); +/* `options?.signal` (+1, undefined when absent) and `{ once: true }` (+1), + * for the listener that makes options.signal abort reduce's controller. */ +ScrDyn *scr_stream_op_signal_of(ScrDyn *options); +ScrDyn *scr_stream_op_once_options(void); +/* The operator callbacks' { signal } (the options argument): the signals + * AbortSignal.any combines (`[options?.signal].filter(Boolean)`) and the + * object around the combined signal. +1 each. */ +ScrDyn *scr_stream_op_signal_list(ScrDyn *options); +ScrDyn *scr_stream_op_signal_opts(ScrDyn *signal); /* finished(s, options[, cb]) — Node's eos with its listeners and state * predicates: readable/writable are -1 (the default), 0 or 1; on_error is * options.error !== false; signal is NULL or borrowed (ERR_INVALID_ARG_TYPE @@ -556,6 +579,16 @@ ScrStream *scr_stream_readable_from_web(struct ScrDyn *web, bool object_mode, do * generator and answers its promise; source and the promise's box helper * borrowed): a writable Duplexify whose final waits for the promise. +1. */ ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box); +/* The same for a body that declares the options argument: it takes + * { signal } (signal borrowed), and destroy calls abort(err) (borrowed), + * the AbortController's abort, before it settles (fromAsyncGen's + * ac.abort(err)). */ +ScrStream *scr_stream_duplex_from_gen_signal(ScrClosure *stage, ScrClosure *source, ScrClosure *next, + ScrClosure *read, ScrClosure *ret, ScrClosure *thr, + ScrClosure *abort, struct ScrDyn *signal); +ScrStream *scr_stream_duplex_from_async_fn_signal(ScrClosure *stage, ScrClosure *source, + ScrClosure *box, ScrClosure *abort, + struct ScrDyn *signal); /* Duplex.from over a Readable, a Writable, or a { readable, writable } * pair (duplexify.js's _duplexify): a Duplexify that reads r and writes w * (either NULL in scr_stream_duplexify; streams borrowed). +1. */ diff --git a/packages/runtime/src/scr_stream_ops.c b/packages/runtime/src/scr_stream_ops.c index 75ce416ac..a92b4a0e5 100644 --- a/packages/runtime/src/scr_stream_ops.c +++ b/packages/runtime/src/scr_stream_ops.c @@ -267,20 +267,15 @@ static bool op_aborted(const ScrDyn *signal) { return r; } -/* The callback argument: a function that does not read the options - * argument. `max` is how many parameters Node's call passes before the - * options. False with the error pending. */ +/* The callback argument: validateFunction. False with the error pending. + * (The options argument the callback may read is the operator's + * cb_opts.) */ static bool op_check_fn(const ScrDyn *fn, const char *name, uint32_t max) { + (void)max; if (!fn || !scr_dyn_is_callable(fn)) { scr_dyn_arg_type_fail(name, "of type function", fn ? fn : scr_dyn_undefined()); return false; } - if (fn->kind == SCR_DYN_FUNC && fn->v.fn.arity > max) { - static const char m[] = - "stream operator callbacks that read the options argument ({ signal }) are not supported yet"; - scr_throw_error_msg(SCR_ERR_ERROR, m, sizeof m - 1); - return false; - } return true; } @@ -610,8 +605,9 @@ enum { MODE_MAP, MODE_FILTER, MODE_FOREACH, MODE_EVERY }; typedef struct { OpGen g; OpSrc src; - ScrDyn *fn; /* +1 */ - ScrDyn *signal; /* +1 or NULL */ + ScrDyn *fn; /* +1 */ + ScrDyn *signal; /* +1 or NULL */ + ScrDyn *cb_opts; /* +1: the { signal } fn takes as its second argument, or NULL */ int mode; double concurrency, hwm; ScrDyn **queue; /* +1 entries: dyn promises or kEof */ @@ -628,6 +624,7 @@ static void op_map_body_drop(OpGen *g) { src_release(&m->src); scr_dyn_release(m->fn); if (m->signal) scr_dyn_release(m->signal); + if (m->cb_opts) scr_dyn_release(m->cb_opts); for (size_t i = m->qh; i < m->qn; i++) scr_dyn_release(m->queue[i]); free(m->queue); if (m->next_p) scr_promise_release(m->next_p); @@ -734,9 +731,11 @@ static void op_wrap_k(void *self, ScrDyn *v) { } /* Calls fn(chunk): +1 result, or NULL with the exception pending. */ -static ScrDyn *op_call(ScrDyn *fn, ScrDyn *chunk) { - ScrDyn *args[1] = {chunk}; - return scr_dyn_call(fn, args, 1, "fn"); +/* Calls fn(chunk, signalOpt) — the second argument when the callback + * declares it. +1 result, or NULL with the exception pending. */ +static ScrDyn *op_call(OpMap *m, ScrDyn *chunk) { + ScrDyn *args[2] = {chunk, m->cb_opts}; + return scr_dyn_call(m->fn, args, m->cb_opts ? 2 : 1, "fn"); } /* An async wrapper body: `await fn(chunk)` then its transform. The @@ -744,7 +743,7 @@ static ScrDyn *op_call(ScrDyn *fn, ScrDyn *chunk) { static ScrPromise *op_wrap_call(OpMap *m, ScrDyn *chunk, int mode, int stage) { OpWrap *w = op_wrap_new(mode, chunk, stage); ScrPromise *result = scr_promise_retain(w->p); - ScrDyn *r = op_call(m->fn, chunk); + ScrDyn *r = op_call(m, chunk); if (!r) { scr_promise_reject_pending(w->p); } else { @@ -772,7 +771,7 @@ static ScrPromise *op_every_call(OpMap *m, ScrDyn *chunk) { static ScrDyn *map_item(OpMap *m, ScrDyn *chunk) { ScrPromise *p; if (m->mode == MODE_MAP) { - ScrDyn *r = op_call(m->fn, chunk); + ScrDyn *r = op_call(m, chunk); if (!r) { ScrCaught *c = scr_exc_take(); ScrDyn *d = op_rejected_dyn(c); @@ -981,7 +980,7 @@ static void map_thr(OpGen *g, ScrCaught *c) { } static OpMap *map_new(ScrStream *s, ScrDyn *fn, int mode, double concurrency, double hwm, - ScrDyn *signal) { + ScrDyn *signal, ScrDyn *cb_opts) { OpMap *m = op_alloc(sizeof *m); gen_init(&m->g, &op_map_body_drop); m->g.resume = &map_resume; @@ -990,6 +989,7 @@ static OpMap *map_new(ScrStream *s, ScrDyn *fn, int mode, double concurrency, do m->src.s = scr_stream_retain(s); m->fn = scr_dyn_retain(fn); m->signal = signal ? scr_dyn_retain(signal) : NULL; + m->cb_opts = cb_opts && cb_opts->kind != SCR_DYN_UNDEF ? scr_dyn_retain(cb_opts) : NULL; m->mode = mode; m->concurrency = concurrency; m->hwm = hwm; @@ -1137,27 +1137,137 @@ static void slice_thr(OpGen *g, ScrCaught *c) { * goes through the async-from-sync iterator: each step resolves after the * element's PromiseResolve job, and yield* awaits that result before * yielding without a further await. */ +/* yield* over the delegate: an array or string (its elements, through + * the async-from-sync iterator), another iterable's iterator (FLAT_SYNC + * through the async-from-sync iterator, FLAT_ASYNC awaiting next()), or a + * native Readable (FLAT_STREAM, its async iterator). */ +enum { FLAT_NONE, FLAT_SYNC, FLAT_ASYNC, FLAT_STREAM, FLAT_GEN_SYNC, FLAT_GEN_ASYNC }; + +typedef ScrDyn *(*FromSyncFn)(ScrClosure *, ScrGen *); +typedef ScrDyn *(*FromSyncThrowFn)(ScrClosure *, ScrGen *, ScrDyn *); +typedef ScrPromise *(*FromAsyncFn)(ScrClosure *, ScrGen *); +typedef ScrPromise *(*FromAsyncThrowFn)(ScrClosure *, ScrGen *, ScrDyn *); +typedef ScrDyn *(*FromReadFn)(ScrClosure *, void *); +static void op_await_gen(ScrPromise *p, ScrClosure *read, OpCont k, void *self); + typedef struct { OpGen g; OpSrc values; /* the map generator */ ScrDyn *inner; /* +1: the array or string being delegated to, or NULL */ size_t idx; /* element index, or the string's byte offset */ ScrPromise *step; /* the async-from-sync iterator result being awaited */ + ScrDyn *iter; /* +1: the delegate's iterator (FLAT_SYNC / FLAT_ASYNC), or NULL */ + ScrStream *stream; /* +1: the delegate stream (FLAT_STREAM), or NULL */ + ScrGen *gen; /* +1: a compiled generator delegate (FLAT_GEN_*), or NULL */ + /* The compiler's helpers over the callback's generator type (+1 each, + * or NULL): next/return/throw, and read for an async generator. */ + ScrClosure *gnext, *gread, *gret, *gthr; + bool gen_async; + int iter_kind; bool started; } OpFlat; +/* A delegate's member (+1), read as the IR's keyed read does: a compiled + * value (a generator, its result record) through its typed ref, anything + * else through the iterator-result read. NULL with the error pending. */ +static ScrDyn *flat_get(const ScrDyn *v, const char *key) { + if (v->kind == SCR_DYN_TYPED_REF) { + ScrStr *k = scr_str_new(key, strlen(key)); + ScrDyn *r = scr_dyn_typed_ref_key_get(v, k); + scr_str_release(k); + return r; + } + return scr_dyn_iterator_result_get(v, key, strlen(key)); +} + +/* iterator[name](...args) with the iterator as this: +1, or NULL with the + * error pending. */ +static ScrDyn *flat_call(const ScrDyn *iter, const char *name, ScrDyn *const *args, size_t argc) { + ScrDyn *m = flat_get(iter, name); + if (!m) return NULL; + scr_dyn_this_push_dyn(iter); + ScrDyn *r = scr_dyn_call(m, args, argc, name); + scr_dyn_this_pop(); + scr_dyn_release(m); + return r; +} + +/* `const { value, done } = result`: +1 value with *done set, or NULL with + * the error pending. */ +static ScrDyn *flat_result(const ScrDyn *r, bool *done) { + if (r->kind == SCR_DYN_UNDEF || r->kind == SCR_DYN_NULL || r->kind == SCR_DYN_NUM || + r->kind == SCR_DYN_STR || r->kind == SCR_DYN_BOOL) { + ScrDyn *checked = scr_dyn_iterator_result((ScrDyn *)r); /* throws V8's TypeError */ + if (checked) scr_dyn_release(checked); + return NULL; + } + ScrDyn *d = flat_get(r, "done"); + if (!d) return NULL; + *done = scr_dyn_truthy(d); + scr_dyn_release(d); + return flat_get(r, "value"); +} + +static void flat_drop_delegate(OpFlat *f) { + if (f->inner) scr_dyn_release(f->inner); + if (f->iter) scr_dyn_release(f->iter); + if (f->stream) scr_stream_release(f->stream); + if (f->gen) scr_gen_release(f->gen); + f->inner = f->iter = NULL; + f->stream = NULL; + f->gen = NULL; + f->iter_kind = FLAT_NONE; +} + static void op_flat_body_drop(OpGen *g) { OpFlat *f = (OpFlat *)g; src_release(&f->values); - if (f->inner) scr_dyn_release(f->inner); + flat_drop_delegate(f); if (f->step) scr_promise_release(f->step); + if (f->gnext) scr_closure_release(f->gnext); + if (f->gread) scr_closure_release(f->gread); + if (f->gret) scr_closure_release(f->gret); + if (f->gthr) scr_closure_release(f->gthr); free(f); } static void flat_next_value(OpFlat *f); -/* The next element (+1), or NULL when the delegate is exhausted. */ -static ScrDyn *flat_take_element(OpFlat *f) { +/* The next element (+1), or NULL when the delegate is exhausted; for a + * sync iterator NULL with *failed set when next() threw (pending). */ +static ScrDyn *flat_take_element(OpFlat *f, bool *failed) { + *failed = false; + if (f->iter_kind == FLAT_GEN_SYNC) { + ScrDyn *v = ((FromSyncFn)f->gnext->fn)(f->gnext, scr_gen_retain(f->gen)); + if (!v) { + *failed = true; + return NULL; + } + if (v == op_k_done) { + scr_dyn_release(v); + return NULL; + } + return v; + } + if (f->iter_kind == FLAT_SYNC) { + ScrDyn *r = flat_call(f->iter, "next", NULL, 0); + if (!r) { + *failed = true; + return NULL; + } + bool done = false; + ScrDyn *v = flat_result(r, &done); + scr_dyn_release(r); + if (!v) { + *failed = true; + return NULL; + } + if (done) { + scr_dyn_release(v); + return NULL; + } + return v; + } ScrDyn *in = f->inner; if (in->kind == SCR_DYN_ARR) { if (f->idx >= in->v.arr.len) return NULL; @@ -1185,10 +1295,15 @@ static void flat_step_k(void *self, ScrDyn *v) { scr_promise_release(f->step); f->step = NULL; } - if (v == op_k_done || (v && v == op_k_eof)) { + if (!v) { + /* the delegate's next() threw: it leaves the for await */ + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + if (v == op_k_done || v == op_k_eof) { scr_dyn_release(v); - scr_dyn_release(f->inner); - f->inner = NULL; + flat_drop_delegate(f); flat_next_value(f); return; } @@ -1196,6 +1311,33 @@ static void flat_step_k(void *self, ScrDyn *v) { scr_dyn_release(v); } +/* Await(innerResult) for an async delegate: done ends it, else its value + * is yielded as it is. */ +static void flat_iter_k(void *self, ScrDyn *v) { + OpFlat *f = self; + if (!v) { + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + bool done = false; + ScrDyn *value = flat_result(v, &done); + scr_dyn_release(v); + if (!value) { + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + if (done) { + scr_dyn_release(value); + flat_drop_delegate(f); + flat_next_value(f); + return; + } + gen_yield_now(&f->g, value); + scr_dyn_release(value); +} + /* The async-from-sync continuation: PromiseResolve(value) then the result * promise settles; yield* awaits that promise. */ typedef struct { @@ -1220,8 +1362,56 @@ static void op_flat_wrap_k(void *self, ScrDyn *v) { scr_dyn_release(v); } +/* The awaited resume of an async compiled generator delegate, read to its + * value or the done marker: done ends it, else the value is yielded. */ +static void flat_gen_k(void *self, ScrDyn *v) { + OpFlat *f = self; + if (!v) { + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + if (v == op_k_done) { + scr_dyn_release(v); + flat_drop_delegate(f); + flat_next_value(f); + return; + } + gen_yield_now(&f->g, v); + scr_dyn_release(v); +} + static void flat_delegate_step(OpFlat *f) { - ScrDyn *e = flat_take_element(f); + if (f->iter_kind == FLAT_GEN_ASYNC) { + ScrPromise *p = ((FromAsyncFn)f->gnext->fn)(f->gnext, scr_gen_retain(f->gen)); + op_await_gen(p, f->gread, &flat_gen_k, f); + scr_promise_release(p); + return; + } + if (f->iter_kind == FLAT_ASYNC) { + ScrDyn *r = flat_call(f->iter, "next", NULL, 0); + if (!r) { + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + op_await_value(r, &flat_iter_k, f); + scr_dyn_release(r); + return; + } + if (f->iter_kind == FLAT_STREAM) { + ScrPromise *p = scr_stream_next_result_dyn(f->stream); + op_await(p, &flat_iter_k, f); + scr_promise_release(p); + return; + } + bool failed = false; + ScrDyn *e = flat_take_element(f, &failed); + if (failed) { + flat_drop_delegate(f); + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } OpFlatWrap *w = op_alloc(sizeof *w); w->base.rc = 1; w->base.drop = &op_flat_wrap_drop; @@ -1247,15 +1437,63 @@ static void flat_val_k(void *self, ScrDyn *v) { gen_complete(&f->g, NULL); return; } - if (v->kind != SCR_DYN_ARR && v->kind != SCR_DYN_STR) { - scr_dyn_release(v); - ScrCaught *c = op_take_error( - op_unsupported("flatMap callback results other than arrays or strings")); - gen_close_source(&f->g, &f->values, c); + if (v->kind == SCR_DYN_ARR || v->kind == SCR_DYN_STR) { + f->inner = v; + f->idx = 0; + flat_delegate_step(f); return; } - f->inner = v; - f->idx = 0; + ScrStream *stream = scr_stream_native_readable_of_dyn(v); + if (stream) { + scr_dyn_release(v); + f->stream = stream; + f->iter_kind = FLAT_STREAM; + flat_delegate_step(f); + return; + } + /* a compiled generator of the callback's declared type: stepped through + * the compiler's helpers */ + if (f->gnext && v->kind == SCR_DYN_TYPED_REF && v->v.typed_ref.type_key && + !strncmp(v->v.typed_ref.type_key, f->gen_async ? "async-generator<" : "generator<", + f->gen_async ? 16 : 10)) { + f->gen = scr_dyn_typed_ref_unbox(v); + scr_dyn_release(v); + f->iter_kind = f->gen_async ? FLAT_GEN_ASYNC : FLAT_GEN_SYNC; + flat_delegate_step(f); + return; + } + /* yield* val: GetIterator(val, async); a compiled generator is its own + * iterator */ + int kind = 0; + ScrDyn *iter = NULL; + if (v->kind == SCR_DYN_TYPED_REF && v->v.typed_ref.type_key && + (!strncmp(v->v.typed_ref.type_key, "async-generator<", 16) || + !strncmp(v->v.typed_ref.type_key, "generator<", 10))) { + kind = v->v.typed_ref.type_key[0] == 'a' ? 2 : 1; + iter = scr_dyn_retain(v); + } else { + iter = scr_dyn_open_iterator(v, &kind, true); + } + if (!iter && kind == 0 && !scr_exc_pending()) { + /* V8's GetIterator over a value without either method */ + if (v->kind == SCR_DYN_NULL || v->kind == SCR_DYN_UNDEF) { + char m[96]; + int n = snprintf(m, sizeof m, + "Cannot read properties of %s (reading 'Symbol(Symbol.asyncIterator)')", + v->kind == SCR_DYN_NULL ? "null" : "undefined"); + scr_throw_error_msg(SCR_ERR_TYPE, m, (size_t)n); + } else { + static const char m[] = "undefined is not a function"; + scr_throw_error_msg(SCR_ERR_TYPE, m, sizeof m - 1); + } + } + scr_dyn_release(v); + if (!iter) { + gen_close_source(&f->g, &f->values, scr_exc_take()); + return; + } + f->iter = iter; + f->iter_kind = kind == 2 ? FLAT_ASYNC : FLAT_SYNC; flat_delegate_step(f); } @@ -1300,19 +1538,85 @@ static void op_flat_ret_k(void *self, ScrDyn *v) { op_await_value(scr_dyn_undefined(), &op_flat_ret_k, r); return; } - if (r->f->inner) { - scr_dyn_release(r->f->inner); - r->f->inner = NULL; - } + flat_drop_delegate(r->f); gen_close_source(&r->f->g, &r->f->values, NULL); } +/* Whether the delegate's iterator has a callable `name` method. */ +static bool flat_iter_has(OpFlat *f, const char *name) { + if (!f->iter) return false; + ScrDyn *m = flat_get(f->iter, name); + if (!m) { + scr_exc_clear(); + return false; + } + bool callable = scr_dyn_is_callable(m); + scr_dyn_release(m); + return callable; +} + static void flat_ret(OpGen *g, ScrDyn *v) { (void)v; + OpFlat *f = (OpFlat *)g; OpFlatRet *r = op_alloc(sizeof *r); r->base.rc = 1; r->base.drop = &op_flat_ret_drop; - r->f = op_ref((OpFlat *)g); + r->f = op_ref(f); + if (f->iter_kind == FLAT_GEN_ASYNC) { + /* the generator's return(): awaited, then the value */ + ScrPromise *p = ((FromAsyncFn)f->gret->fn)(f->gret, scr_gen_retain(f->gen)); + op_await_gen(p, f->gread, &op_flat_ret_k, r); + scr_promise_release(p); + op_unref(r); + return; + } + if (f->iter_kind == FLAT_GEN_SYNC) { + ScrDyn *res = ((FromSyncFn)f->gret->fn)(f->gret, scr_gen_retain(f->gen)); + if (!res) { + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_exc_take()); + op_unref(r); + return; + } + scr_dyn_release(res); + op_await_value(scr_dyn_undefined(), &op_flat_ret_k, r); + op_unref(r); + return; + } + if (f->iter_kind == FLAT_ASYNC) { + /* yield*'s return: Await(iterator.return(value)), then the value */ + if (flat_iter_has(f, "return")) { + ScrDyn *arg = scr_dyn_undefined(); + ScrDyn *res = flat_call(f->iter, "return", &arg, 1); + if (!res) { + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_exc_take()); + op_unref(r); + return; + } + op_await_value(res, &op_flat_ret_k, r); + scr_dyn_release(res); + } else { + r->stage = 1; + op_await_value(scr_dyn_undefined(), &op_flat_ret_k, r); + } + op_unref(r); + return; + } + if (f->iter_kind == FLAT_SYNC && flat_iter_has(f, "return")) { + /* the async-from-sync iterator's return() calls the sync one */ + ScrDyn *arg = scr_dyn_undefined(); + ScrDyn *res = flat_call(f->iter, "return", &arg, 1); + if (!res) { + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_exc_take()); + op_unref(r); + return; + } + scr_dyn_release(res); + } else if (f->iter_kind == FLAT_STREAM) { + scr_stream_release(scr_stream_iterator_close(f->stream)); + } op_await_value(scr_dyn_undefined(), &op_flat_ret_k, r); op_unref(r); } @@ -1321,12 +1625,72 @@ static void flat_ret(OpGen *g, ScrDyn *v) { * yield* rejects with Node's TypeError, which leaves the for await and * closes map's generator. */ static void flat_thr(OpGen *g, ScrCaught *c) { - (void)c; OpFlat *f = (OpFlat *)g; - if (f->inner) { - scr_dyn_release(f->inner); - f->inner = NULL; + if (f->iter_kind == FLAT_GEN_ASYNC || f->iter_kind == FLAT_GEN_SYNC) { + /* the generator's throw(error) answers the next step */ + ScrDyn *err = scr_caught_to_dyn(c); + if (f->iter_kind == FLAT_GEN_ASYNC) { + ScrPromise *p = ((FromAsyncThrowFn)f->gthr->fn)(f->gthr, scr_gen_retain(f->gen), err); + op_await_gen(p, f->gread, &flat_gen_k, f); + scr_promise_release(p); + return; + } + ScrDyn *v = ((FromSyncThrowFn)f->gthr->fn)(f->gthr, scr_gen_retain(f->gen), err); + if (!v) { + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_exc_take()); + return; + } + if (v == op_k_done) { + scr_dyn_release(v); + flat_drop_delegate(f); + flat_next_value(f); + return; + } + op_await_value(v, &flat_step_k, f); + scr_dyn_release(v); + return; } + if ((f->iter_kind == FLAT_ASYNC || f->iter_kind == FLAT_SYNC) && flat_iter_has(f, "throw")) { + /* yield*'s throw: the delegate's throw(error) answers the next step */ + ScrDyn *err = scr_caught_to_dyn(c); + ScrDyn *res = flat_call(f->iter, "throw", &err, 1); + scr_dyn_release(err); + if (!res) { + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_exc_take()); + return; + } + if (f->iter_kind == FLAT_ASYNC) { + op_await_value(res, &flat_iter_k, f); + scr_dyn_release(res); + return; + } + ScrPromise *p = op_resolved(res); + scr_dyn_release(res); + op_await(p, &flat_iter_k, f); + scr_promise_release(p); + return; + } + if (f->iter_kind == FLAT_STREAM) { + /* the stream's iterator rethrows after destroying the stream */ + ScrDyn *err = scr_caught_to_dyn(c); + ScrError *e = err ? scr_error_from_dyn(err) : NULL; + if (scr_exc_pending()) scr_exc_clear(); + if (err) scr_dyn_release(err); + scr_stream_release(scr_stream_destroy(f->stream, e)); + if (e) scr_error_release(e); + flat_drop_delegate(f); + gen_close_source(g, &f->values, scr_caught_retain(c)); + return; + } + if (f->iter_kind != FLAT_NONE && flat_iter_has(f, "return")) { + /* no throw method: the iterator closes first */ + ScrDyn *res = flat_call(f->iter, "return", NULL, 0); + if (res) scr_dyn_release(res); + else scr_exc_clear(); + } + flat_drop_delegate(f); ScrError *e = op_coded_error(SCR_ERR_TYPE, NULL, "The iterator does not provide a 'throw' method."); gen_close_source(g, &f->values, op_take_error(e)); } @@ -1370,12 +1734,6 @@ static void op_from_drop(OpObj *o) { /* The helper ABIs: the closure first, then the declared parameters, which * the callee owns. */ -typedef ScrDyn *(*FromSyncFn)(ScrClosure *, ScrGen *); -typedef ScrDyn *(*FromSyncThrowFn)(ScrClosure *, ScrGen *, ScrDyn *); -typedef ScrPromise *(*FromAsyncFn)(ScrClosure *, ScrGen *); -typedef ScrPromise *(*FromAsyncThrowFn)(ScrClosure *, ScrGen *, ScrDyn *); -typedef ScrDyn *(*FromReadFn)(ScrClosure *, void *); - /* `await p` over an async generator's resume promise: the IteratorResult * record goes through `read` (the value, or the done marker) before k. * Awaiting the typed promise itself keeps from.js's single await. */ @@ -2052,6 +2410,7 @@ struct OpChan { ScrPromise *resolve; /* +1: the deferred a write or the final settles, NULL once settled */ int settled; /* CHAN_*: what settled the deferred, for ack() */ bool aborted; /* destroy ran (fromAsyncGen's ac.abort()) */ + ScrClosure *abort; /* +1: the body's AbortController's abort(), or NULL */ /* An async function body: its promise's then() (duplexify.js) settles * `promise` there; the final waits for it. */ bool finalized; /* final() has started */ @@ -2066,6 +2425,7 @@ static SCR_TL OpChan *op_chan_live = NULL; static void op_chan_drop(OpObj *o) { OpChan *c = (OpChan *)o; if (c->body_err) scr_error_release(c->body_err); + if (c->abort) scr_closure_release(c->abort); if (c->d) scr_stream_release(c->d); if (c->promise) scr_promise_release(c->promise); if (c->resolve) scr_promise_release(c->resolve); @@ -2091,6 +2451,8 @@ static void op_chan_retire(OpChan *c) { c->live_prev = c->live_next = NULL; ScrStream *d = c->d; c->d = NULL; + if (c->abort) scr_closure_release(c->abort); + c->abort = NULL; if (c->promise) scr_promise_release(c->promise); if (c->resolve) scr_promise_release(c->resolve); c->promise = c->resolve = NULL; @@ -2233,6 +2595,21 @@ static void op_chan_final_inv(ScrClosure *cb, ScrStream *s) { op_unref(c); } +/* fromAsyncGen's ac.abort(err): the body's signal fires its abort + * listeners once (a body that declares the options argument); without an + * error the reason is null (destroy passes `err || null`). */ +static void op_chan_abort(OpChan *c, ScrError *err) { + c->aborted = true; + ScrClosure *abort = c->abort; + if (!abort) return; + c->abort = NULL; + ScrDyn *reason = err ? scr_dyn_from_error(err) : NULL; + if (!reason) reason = scr_dyn_new_null(); + ((void (*)(ScrClosure *, ScrDyn *))abort->fn)(abort, reason); /* reason moves in */ + if (scr_exc_pending()) scr_exc_clear(); + scr_closure_release(abort); +} + /* destroy: fromAsyncGen's (ac.abort(); a source waiting for a write or * the final is unblocked with the done marker and a no-op callback; * cb(err)), then from()'s close of the function's iterator. caps: the @@ -2240,7 +2617,7 @@ static void op_chan_final_inv(ScrClosure *cb, ScrStream *s) { static void op_dup_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { OpChan *c = scr_box_get_ref(cb->caps[0]); ScrClosure *from = scr_box_get_ref(cb->caps[1]); - c->aborted = true; + op_chan_abort(c, err); if (c->resolve) op_chan_settle(c, scr_stream_from_done(), CHAN_ABORT); op_chan_retire(c); op_unref(c); @@ -2250,14 +2627,11 @@ static void op_dup_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { typedef ScrGen *(*OpSourceFn)(ScrClosure *, ScrClosure *, ScrClosure *); typedef ScrGen *(*OpStageFn)(ScrClosure *, ScrGen *); +typedef ScrGen *(*OpStageOptsFn)(ScrClosure *, ScrGen *, ScrDyn *); -/* Duplex.from(fn) for an async generator function: the compiled stage - * (borrowed) takes the source generator and answers its own; source - * (borrowed) makes the source generator from next() and ack(); the - * compiler's helpers over the stage's generator are borrowed. +1, or - * NULL with the function's throw pending. */ -ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, ScrClosure *next, - ScrClosure *read, ScrClosure *ret, ScrClosure *thr) { +/* A new channel; with a signal (borrowed), the body's AbortController's + * abort() (borrowed) is kept for destroy and *opts is { signal } (+1). */ +static OpChan *op_chan_new(ScrClosure *abort, ScrDyn *signal, ScrDyn **opts) { op_sentinels_init(); static SCR_TL bool registered = false; if (!registered) { @@ -2267,10 +2641,35 @@ ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, Scr OpChan *c = op_alloc(sizeof *c); c->base.rc = 1; c->base.drop = &op_chan_drop; + *opts = NULL; + if (signal) { + c->abort = scr_closure_retain(abort); + *opts = scr_dyn_new_obj(); + scr_dyn_obj_set(*opts, "signal", 6, scr_dyn_retain(signal)); + } op_chan_defer(c); + return c; +} + +/* Duplex.from(fn) for an async generator function: the compiled stage + * (borrowed) takes the source generator and answers its own; source + * (borrowed) makes the source generator from next() and ack(); the + * compiler's helpers over the stage's generator are borrowed. With a + * signal, the stage also takes { signal }. +1, or NULL with the + * function's throw pending. */ +static ScrStream *op_duplex_from_gen(ScrClosure *stage, ScrClosure *source, ScrClosure *next, + ScrClosure *read, ScrClosure *ret, ScrClosure *thr, + ScrClosure *abort, ScrDyn *signal) { + ScrDyn *opts; + OpChan *c = op_chan_new(abort, signal, &opts); ScrGen *src = ((OpSourceFn)source->fn)(source, op_chan_closure((void *)&op_chan_next_inv, c), op_chan_closure((void *)&op_chan_ack_inv, c)); - ScrGen *g = scr_exc_pending() ? NULL : ((OpStageFn)stage->fn)(stage, src); /* src moves in */ + ScrGen *g = NULL; /* src and opts move in */ + if (scr_exc_pending()) { + if (opts) scr_dyn_release(opts); + } else { + g = opts ? ((OpStageOptsFn)stage->fn)(stage, src, opts) : ((OpStageFn)stage->fn)(stage, src); + } if (scr_exc_pending()) { if (g) scr_gen_release(g); op_unref(c); @@ -2307,6 +2706,17 @@ ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, Scr return d; } +ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, ScrClosure *next, + ScrClosure *read, ScrClosure *ret, ScrClosure *thr) { + return op_duplex_from_gen(stage, source, next, read, ret, thr, NULL, NULL); +} + +ScrStream *scr_stream_duplex_from_gen_signal(ScrClosure *stage, ScrClosure *source, ScrClosure *next, + ScrClosure *read, ScrClosure *ret, ScrClosure *thr, + ScrClosure *abort, ScrDyn *signal) { + return op_duplex_from_gen(stage, source, next, read, ret, thr, abort, signal); +} + /* ── Duplex.from(asyncFunction) ── duplexify.js: the function's promise * must settle to a nullish value (else ERR_INVALID_RETURN_VALUE when the * final awaits it), a return before the input was consumed destroys the @@ -2314,6 +2724,7 @@ ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, Scr * writable only, its final waiting for the promise. */ typedef ScrPromise *(*OpAsyncStageFn)(ScrClosure *, ScrGen *); +typedef ScrPromise *(*OpAsyncStageOptsFn)(ScrClosure *, ScrGen *, ScrDyn *); typedef ScrDyn *(*OpBoxFn)(ScrClosure *, ScrPromise *); typedef struct { @@ -2376,7 +2787,7 @@ static void op_chan_final_async_inv(ScrClosure *cb, ScrStream *s) { /* fromAsyncGen's destroy alone (no from() iterator to close) */ static void op_chan_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { OpChan *c = op_chan_of(cb); - c->aborted = true; + op_chan_abort(c, err); if (c->resolve) op_chan_settle(c, scr_stream_from_done(), CHAN_ABORT); op_chan_retire(c); op_unref(c); @@ -2384,23 +2795,22 @@ static void op_chan_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { } /* Duplex.from(fn) for an async function: the compiled stage (borrowed) - * takes the source generator and answers its promise; source and box - * (promise → dyn promise) borrowed. +1, or NULL with the function's throw - * pending. */ -ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box) { - op_sentinels_init(); - static SCR_TL bool registered = false; - if (!registered) { - registered = true; - scr_atexit(op_chan_atexit); - } - OpChan *c = op_alloc(sizeof *c); - c->base.rc = 1; - c->base.drop = &op_chan_drop; - op_chan_defer(c); + * takes the source generator (and, with a signal, { signal }) and answers + * its promise; source and box (promise → dyn promise) borrowed. +1, or + * NULL with the function's throw pending. */ +static ScrStream *op_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box, + ScrClosure *abort, ScrDyn *signal) { + ScrDyn *opts; + OpChan *c = op_chan_new(abort, signal, &opts); ScrGen *src = ((OpSourceFn)source->fn)(source, op_chan_closure((void *)&op_chan_next_inv, c), op_chan_closure((void *)&op_chan_ack_inv, c)); - ScrPromise *p = scr_exc_pending() ? NULL : ((OpAsyncStageFn)stage->fn)(stage, src); + ScrPromise *p = NULL; /* src and opts move in */ + if (scr_exc_pending()) { + if (opts) scr_dyn_release(opts); + } else { + p = opts ? ((OpAsyncStageOptsFn)stage->fn)(stage, src, opts) + : ((OpAsyncStageFn)stage->fn)(stage, src); + } if (scr_exc_pending()) { if (p) scr_promise_release(p); op_unref(c); @@ -2427,6 +2837,16 @@ ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source return d; } +ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box) { + return op_duplex_from_async_fn(stage, source, box, NULL, NULL); +} + +ScrStream *scr_stream_duplex_from_async_fn_signal(ScrClosure *stage, ScrClosure *source, + ScrClosure *box, ScrClosure *abort, + ScrDyn *signal) { + return op_duplex_from_async_fn(stage, source, box, abort, signal); +} + /* ── the promise operators ─────────────────────────────────────────── */ enum { PK_DYN, PK_F64, PK_BOOL, PK_STR, PK_VOID }; @@ -2439,6 +2859,8 @@ typedef struct { ScrDyn *fn; /* +1 or NULL */ ScrDyn *signal; /* +1 or NULL */ ScrDyn *acc; /* toArray's array / reduce's accumulator (+1) */ + ScrClosure *ctl; /* +1: reduce's ac.abort(), run when it ends, or NULL */ + ScrDyn *cb_opts; /* +1: the { signal } the reducer takes as its third argument, or NULL */ int op; int kind; /* the promise payload (PK_*) */ bool has_init, got_any, negate; @@ -2453,9 +2875,21 @@ static void op_fn_drop(OpObj *o) { if (a->fn) scr_dyn_release(a->fn); if (a->signal) scr_dyn_release(a->signal); if (a->acc) scr_dyn_release(a->acc); + if (a->ctl) scr_closure_release(a->ctl); + if (a->cb_opts) scr_dyn_release(a->cb_opts); free(a); } +/* reduce's `finally { ac.abort(); }`, before its promise settles. */ +static void op_fn_abort_ctl(OpFn *a) { + if (!a->ctl) return; + ScrClosure *ctl = a->ctl; + a->ctl = NULL; + ((void (*)(ScrClosure *))ctl->fn)(ctl); + if (scr_exc_pending()) scr_exc_clear(); + scr_closure_release(ctl); +} + static OpFn *op_fn_new(int op, int kind) { OpFn *a = op_alloc(sizeof *a); a->base.rc = 1; @@ -2469,6 +2903,7 @@ static OpFn *op_fn_new(int op, int kind) { /* Settles the operator's promise with v converted to its payload kind (a * value of the wrong kind rejects with the checked conversion's error). */ static void op_fn_resolve(OpFn *a, ScrDyn *v /*borrowed*/) { + op_fn_abort_ctl(a); switch (a->kind) { case PK_VOID: scr_promise_fulfill_void(a->p); @@ -2493,7 +2928,10 @@ static void op_fn_resolve(OpFn *a, ScrDyn *v /*borrowed*/) { scr_promise_reject_pending(a->p); } -static void op_fn_reject(OpFn *a, const ScrCaught *c) { op_reject(a->p, c); } +static void op_fn_reject(OpFn *a, const ScrCaught *c) { + op_fn_abort_ctl(a); + op_reject(a->p, c); +} static void op_fn_next(OpFn *a); @@ -2634,8 +3072,8 @@ static void op_fn_val_k(void *self, ScrDyn *v) { op_fn_next(a); return; } - ScrDyn *args[2] = {a->acc, v}; - ScrDyn *r = scr_dyn_call(a->fn, args, 2, "reducer"); + ScrDyn *args[3] = {a->acc, v, a->cb_opts}; + ScrDyn *r = scr_dyn_call(a->fn, args, a->cb_opts ? 3 : 2, "reducer"); scr_dyn_release(v); if (!r) { op_fn_close(a, scr_exc_take(), NULL); @@ -2717,7 +3155,7 @@ static bool op_map_options(const ScrDyn *options, double *concurrency, double *h /* map(fn, options) with fn already checked: +1 generator, or NULL with * the error pending. */ -static OpMap *op_map_make(ScrStream *s, ScrDyn *fn, ScrDyn *options, int mode) { +static OpMap *op_map_make(ScrStream *s, ScrDyn *fn, ScrDyn *options, int mode, ScrDyn *cb_opts) { op_sentinels_init(); if (!op_check_options(options)) return NULL; bool ok; @@ -2728,26 +3166,28 @@ static OpMap *op_map_make(ScrStream *s, ScrDyn *fn, ScrDyn *options, int mode) { if (signal) scr_dyn_release(signal); return NULL; } - OpMap *m = map_new(s, fn, mode, concurrency, hwm, signal); + OpMap *m = map_new(s, fn, mode, concurrency, hwm, signal, cb_opts); if (signal) scr_dyn_release(signal); return m; } -ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options) { +ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { if (!op_check_fn(fn, "fn", 1)) return NULL; - OpMap *m = op_map_make(s, fn, options, MODE_MAP); + OpMap *m = op_map_make(s, fn, options, MODE_MAP, cb_opts); return m ? op_from_gen(&m->g) : NULL; } -ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options) { +ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { if (!op_check_fn(fn, "fn", 1)) return NULL; - OpMap *m = op_map_make(s, fn, options, MODE_FILTER); + OpMap *m = op_map_make(s, fn, options, MODE_FILTER, cb_opts); return m ? op_from_gen(&m->g) : NULL; } -ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options) { +static ScrStream *op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts, + ScrClosure *next, ScrClosure *read, ScrClosure *ret, + ScrClosure *thr, bool async_gen) { if (!op_check_fn(fn, "fn", 1)) return NULL; - OpMap *m = op_map_make(s, fn, options, MODE_MAP); + OpMap *m = op_map_make(s, fn, options, MODE_MAP, cb_opts); if (!m) return NULL; OpFlat *f = op_alloc(sizeof *f); gen_init(&f->g, &op_flat_body_drop); @@ -2755,9 +3195,34 @@ ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options) { f->g.ret = &flat_ret; f->g.thr = &flat_thr; f->values.g = &m->g; /* moves */ + if (next) { + f->gnext = scr_closure_retain(next); + f->gread = read ? scr_closure_retain(read) : NULL; + f->gret = scr_closure_retain(ret); + f->gthr = scr_closure_retain(thr); + f->gen_async = async_gen; + } return op_from_gen(&f->g); } +ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { + return op_flat_map(s, fn, options, cb_opts, NULL, NULL, NULL, NULL, false); +} + +/* flatMap whose callback returns a compiled generator: the helpers over + * its type (borrowed) step the generators it yields from. */ +ScrStream *scr_stream_op_flat_map_async_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options, + ScrDyn *cb_opts, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, ScrClosure *thr) { + return op_flat_map(s, fn, options, cb_opts, next, read, ret, thr, true); +} + +ScrStream *scr_stream_op_flat_map_sync_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options, + ScrDyn *cb_opts, ScrClosure *next, ScrClosure *ret, + ScrClosure *thr) { + return op_flat_map(s, fn, options, cb_opts, next, NULL, ret, thr, false); +} + /* drop(number, options) / take(number, options) */ static ScrStream *op_slice(ScrStream *s, ScrDyn *number, ScrDyn *options, bool take) { op_sentinels_init(); @@ -2842,8 +3307,8 @@ static void op_noop_listener(ScrClosure *cb, void *err) { (void)err; } -ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, - ScrDyn *options, double kind) { +static ScrPromise *op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, + ScrDyn *options, double kind, ScrClosure *ctl, ScrDyn *cb_opts) { op_sentinels_init(); OpFn *a = op_fn_new(FN_REDUCE, (int)kind); if (!op_check_fn(fn, "reducer", 2)) return op_fn_reject_now(a); @@ -2859,6 +3324,7 @@ ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool ha ScrDyn *reason = op_get(a->signal, "reason"); ScrError *e = op_abort_error(reason ? reason : scr_dyn_undefined()); if (reason) scr_dyn_release(reason); + /* (thrown before the controller exists) */ /* this.once('error', () => {}) */ ScrStr *name = scr_str_new("error", 5); scr_emitter_release(scr_emitter_on((ScrEmitter *)s, name, @@ -2874,40 +3340,97 @@ ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool ha op_unref(a); return p; } + /* const ac = new AbortController() (options.signal's listener that + * aborts it too is registered by the caller) */ + if (ctl) { + a->ctl = scr_closure_retain(ctl); + a->cb_opts = cb_opts && cb_opts->kind != SCR_DYN_UNDEF ? scr_dyn_retain(cb_opts) : NULL; + } return op_fn_start(a); } +ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, + ScrDyn *options, double kind) { + return op_reduce(s, fn, init, has_init, options, kind, NULL, NULL); +} + +/* reduce with a reducer that reads its { signal }: abort is ac.abort() + * (borrowed), cb_opts `{ signal: ac.signal }`. */ +ScrPromise *scr_stream_op_reduce_signal(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init, + ScrDyn *options, double kind, ScrClosure *abort, + ScrDyn *cb_opts) { + return op_reduce(s, fn, init, has_init, options, kind, abort, cb_opts); +} + +/* `options?.signal`: +1, undefined when absent. */ +ScrDyn *scr_stream_op_signal_of(ScrDyn *options) { + if (!options || options->kind == SCR_DYN_UNDEF || options->kind == SCR_DYN_NULL) + return scr_dyn_retain(scr_dyn_undefined()); + ScrDyn *sig = op_get(options, "signal"); + if (scr_exc_pending()) scr_exc_clear(); + return sig ? sig : scr_dyn_retain(scr_dyn_undefined()); +} + +/* `{ once: true }`: +1. */ +ScrDyn *scr_stream_op_once_options(void) { + ScrDyn *o = scr_dyn_new_obj(); + scr_dyn_obj_set(o, "once", 4, scr_dyn_new_bool(true)); + return o; +} + +/* `[options?.signal].filter(Boolean)` for AbortSignal.any: +1 array. */ +ScrDyn *scr_stream_op_signal_list(ScrDyn *options) { + ScrDyn *list = scr_dyn_new_arr(); + if (options && options->kind != SCR_DYN_UNDEF && options->kind != SCR_DYN_NULL) { + /* A signal that fails validateAbortSignal is left out: the operator + * itself throws Node's ERR_INVALID_ARG_TYPE for it. */ + bool ok; + ScrDyn *sig = op_signal(options, &ok); + if (sig) scr_dyn_arr_push(list, sig); /* moves */ + if (scr_exc_pending()) scr_exc_clear(); + } + return list; +} + +/* `{ signal }`, the callbacks' second argument: +1 object (borrows signal). */ +ScrDyn *scr_stream_op_signal_opts(ScrDyn *signal) { + ScrDyn *o = scr_dyn_new_obj(); + scr_dyn_obj_set(o, "signal", 6, scr_dyn_retain(signal)); + return o; +} + /* some/find/every/forEach: the loop over map's generator (filter's for * some, find and every). */ -static ScrPromise *op_gen_fn(ScrStream *s, ScrDyn *fn, ScrDyn *options, int op, int mode, int kind) { +static ScrPromise *op_gen_fn(ScrStream *s, ScrDyn *fn, ScrDyn *options, int op, int mode, int kind, + ScrDyn *cb_opts) { op_sentinels_init(); OpFn *a = op_fn_new(op, kind); if (!op_check_fn(fn, "fn", 1)) return op_fn_reject_now(a); - OpMap *m = op_map_make(s, fn, options, mode); + OpMap *m = op_map_make(s, fn, options, mode, cb_opts); if (!m) return op_fn_reject_now(a); a->src.g = &m->g; /* moves */ return op_fn_start(a); } -ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options) { - return op_gen_fn(s, fn, options, FN_SOME, MODE_FILTER, PK_BOOL); +ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { + return op_gen_fn(s, fn, options, FN_SOME, MODE_FILTER, PK_BOOL, cb_opts); } -ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options) { - return op_gen_fn(s, fn, options, FN_FIND, MODE_FILTER, PK_DYN); +ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { + return op_gen_fn(s, fn, options, FN_FIND, MODE_FILTER, PK_DYN, cb_opts); } -ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options) { - return op_gen_fn(s, fn, options, FN_FOR_EACH, MODE_FOREACH, PK_VOID); +ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { + return op_gen_fn(s, fn, options, FN_FOR_EACH, MODE_FOREACH, PK_VOID, cb_opts); } /* every(fn, options) = !(await some.call(this, async (...args) => * !(await fn(...args)), options)) */ -ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options) { +ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts) { op_sentinels_init(); OpFn *outer = op_fn_new(FN_EVERY, PK_BOOL); if (!op_check_fn(fn, "fn", 1)) return op_fn_reject_now(outer); - ScrPromise *some = op_gen_fn(s, fn, options, FN_EVERY, MODE_EVERY, PK_DYN); + ScrPromise *some = op_gen_fn(s, fn, options, FN_EVERY, MODE_EVERY, PK_DYN, cb_opts); ScrPromise *p = scr_promise_retain(outer->p); op_await(some, &op_every_k, outer); scr_promise_release(some); diff --git a/tests/corpus/7433-readable-operator-signal.mjs b/tests/corpus/7433-readable-operator-signal.mjs new file mode 100644 index 000000000..38bfbf367 --- /dev/null +++ b/tests/corpus/7433-readable-operator-signal.mjs @@ -0,0 +1,36 @@ +// The Readable operators pass their callbacks { signal } as the second +// argument; map and filter's signal aborts when the options' signal +// does, reduce's signal is aborted once reduce settles, and map may +// return any value. +import { Readable } from "node:stream"; +const log = []; +log.push(JSON.stringify(await Readable.from([1, 2, 3]).map((n) => ({ n, sq: n * n })).toArray())); +log.push(JSON.stringify(await Readable.from(["a", "bb"]).map((s) => s.length).toArray())); +log.push(JSON.stringify(await Readable.from([1, 2]).map((n, { signal }) => `${n}:${signal.aborted}`).toArray())); +log.push(JSON.stringify(await Readable.from([1, 2, 3, 4]).filter(async (n, opts) => n % 2 === 0 && !opts.signal.aborted).toArray())); +log.push(String(await Readable.from([1, 2, 3]).reduce((acc, n, { signal }) => acc + n + (signal.aborted ? 100 : 0), 0))); +let kept; +await Readable.from([1]).reduce((acc, n, { signal }) => { kept = signal; return acc + n; }, 0); +log.push(`reduce signal after ${kept.aborted}`); +let mapKept; +await Readable.from([1]).map((n, { signal }) => { mapKept = signal; return n; }).toArray(); +log.push(`map signal after ${mapKept.aborted}`); +const ac = new AbortController(); +const events = []; +const p = Readable.from([1, 2, 3]).map(async (n, { signal }) => { + signal.addEventListener("abort", () => events.push(`abort ${n}`)); + await new Promise((r) => setTimeout(r, 10)); + return n; +}, { signal: ac.signal }).toArray(); +setTimeout(() => ac.abort(), 5); +try { await p; } catch (e) { log.push(`map aborted ${e.name}`); } +log.push(events.join(",")); +const dyn = [Readable.from([4, 5])][0]; +log.push(JSON.stringify(await dyn.map((n, { signal }) => n + Number(signal.aborted)).toArray())); +let seen = 0; +await Readable.from([5, 6]).forEach((n, { signal }) => { if (signal) seen++; }); +log.push(`forEach ${seen}`); +log.push(String(await Readable.from([1, 2, 3]).some((n, { signal }) => n > 2 && !signal.aborted))); +log.push(String(await Readable.from([1, 2, 3]).find((n, { signal }) => n > 1 && !signal.aborted))); +log.push(String(await Readable.from([1, 2, 3]).every((n, { signal }) => n > 0 && !signal.aborted))); +console.log(log.join("\n")); diff --git a/tests/corpus/7434-readable-operator-values.ts b/tests/corpus/7434-readable-operator-values.ts new file mode 100644 index 000000000..49f7a0fa9 --- /dev/null +++ b/tests/corpus/7434-readable-operator-values.ts @@ -0,0 +1,32 @@ +// Typed Readable operators: map returns numbers and records, flatMap +// takes generator functions and sets, the callbacks may declare the +// { signal } argument, and read() on an object-mode stream answers the +// values pushed. +import { Readable } from "node:stream"; +async function* pair(n: number): AsyncGenerator { + yield n; + yield n * 100; +} +function* letters(s: string): Generator { + for (const c of s) yield c; +} +async function main(): Promise { + const squares = await Readable.from([1, 2, 3]).map((n: number) => n * n).toArray(); + const records = await Readable.from(["a", "bb"]).map((s: string) => ({ s, len: s.length })).toArray(); + const pairs = await Readable.from([1, 2]).flatMap(pair).toArray(); + const chars = await Readable.from(["hi"]).flatMap(letters).toArray(); + const sets = await Readable.from([7]).flatMap((n: number) => new Set([n, n + 1])).toArray(); + const signals = await Readable.from([1]).map((n: number, opts?: { signal?: AbortSignal }) => `${n}:${opts?.signal?.aborted}`).toArray(); + const total = await Readable.from([1, 2]).reduce((acc: number, n: number, opts?: { signal?: AbortSignal }) => acc + n + (opts?.signal?.aborted ? 1 : 0), 0); + console.log(JSON.stringify({ squares, records, pairs, chars, sets, signals, total })); + const r = new Readable({ objectMode: true, read(): void {} }); + r.push({ a: 1 }); + r.push(5); + r.push(null); + r.once("readable", () => { + const first: unknown = r.read(); + const second: unknown = r.read(); + console.log(JSON.stringify(first), second, r.read()); + }); +} +void main(); diff --git a/tests/corpus/7435-readable-flatmap-iterables.mjs b/tests/corpus/7435-readable-flatmap-iterables.mjs new file mode 100644 index 000000000..f0ee79b5a --- /dev/null +++ b/tests/corpus/7435-readable-flatmap-iterables.mjs @@ -0,0 +1,29 @@ +// flatMap over what the callback returns: async and sync generators, a +// Set, a Map, a Readable, and a value that is not iterable (V8's +// TypeError); the generator's steps interleave with process.nextTick as +// in Node, breaking out of the loop runs the inner generator's finally, +// and an inner throw rejects toArray(). +import { Readable } from "node:stream"; +const log = []; +log.push(JSON.stringify(await Readable.from(["x"]).flatMap(async function* (s) { yield s; yield s + s; }).toArray())); +log.push(JSON.stringify(await Readable.from([2]).flatMap(function* (n) { yield n; yield -n; }).toArray())); +log.push(JSON.stringify(await Readable.from([1, 2]).flatMap((n) => new Set([n, n])).toArray())); +log.push(JSON.stringify(await Readable.from(["p"]).flatMap((s) => Readable.from([s, s.toUpperCase()])).toArray())); +log.push(JSON.stringify(await Readable.from([3]).flatMap((n) => new Map([[n, "v"]])).toArray())); +log.push(JSON.stringify(await Readable.from([{ k: 1 }]).flatMap((o) => [o, { k: o.k + 1 }]).toArray())); +try { await Readable.from([1]).flatMap(() => 5).toArray(); } catch (e) { log.push(`${e.name} ${e.message}`); } +try { await Readable.from([1]).flatMap(() => null).toArray(); } catch (e) { log.push(`${e.name} ${e.message}`); } +const order = []; +let n = 0; +const tick = () => { order.push(`t${n}`); if (++n < 5) process.nextTick(tick); }; +process.nextTick(tick); +for await (const v of Readable.from([1]).flatMap(async function* (k) { order.push("gen start"); yield k; order.push("gen mid"); yield k + 1; })) order.push(`v${v}`); +log.push(order.join(" ")); +const closed = []; +const early = Readable.from([1, 2]).flatMap(async function* (k) { try { yield k; yield k * 10; } finally { closed.push(`finally ${k}`); } }); +for await (const v of early) { closed.push(`got ${v}`); break; } +log.push(closed.join(" ")); +const thrown = []; +try { await Readable.from([1]).flatMap(async function* () { yield 1; throw new Error("inner boom"); }).toArray(); } catch (e) { thrown.push(e.message); } +log.push(thrown.join(" ")); +console.log(log.join("\n")); diff --git a/tests/corpus/7436-readable-read-any-value.mjs b/tests/corpus/7436-readable-read-any-value.mjs new file mode 100644 index 000000000..5e19b16c8 --- /dev/null +++ b/tests/corpus/7436-readable-read-any-value.mjs @@ -0,0 +1,17 @@ +// readable.read() answers any value on an object-mode stream, and a +// Buffer or a string (with an encoding) in byte mode. +import { Readable, PassThrough } from "node:stream"; +const r = new Readable({ objectMode: true, read() {} }); +r.push({ a: 1 }); r.push(2); r.push("three"); r.push([4]); r.push(null); +r.once("readable", () => { let v; while ((v = r.read()) !== null) console.log("read", JSON.stringify(v)); }); +const p = new PassThrough({ objectMode: true }); +p.write([1, 2]); +p.end(); +p.once("readable", () => console.log("pt", JSON.stringify(p.read()), p.read())); +const b = new PassThrough(); +b.end("bytes!"); +b.once("readable", () => { const c = b.read(3); console.log("bytes", Buffer.isBuffer(c), String(c), String(b.read())); }); +const e = new PassThrough(); +e.setEncoding("utf8"); +e.end("text"); +e.once("readable", () => { const t = e.read(); console.log("encoded", typeof t, t); }); diff --git a/tests/corpus/7437-duplex-from-signal.mjs b/tests/corpus/7437-duplex-from-signal.mjs new file mode 100644 index 000000000..263225bbb --- /dev/null +++ b/tests/corpus/7437-duplex-from-signal.mjs @@ -0,0 +1,54 @@ +// A Duplex.from body may take { signal } (duplexify.js's fromAsyncGen): +// destroy aborts it with the destroy error, or with an AbortError when +// there is none, and compose passes it to its function stages. +import { Duplex, Readable, compose } from "node:stream"; +import { once } from "node:events"; + +const d = Duplex.from(async function* (source, { signal }) { + console.log("body", typeof signal, signal.aborted); + signal.addEventListener("abort", () => console.log("abort event", signal.aborted, signal.reason)); + for await (const chunk of source) yield String(chunk).toUpperCase(); + console.log("source ended"); +}); +d.on("data", (c) => console.log("data", c)); +d.write("a"); +d.write("b"); +await new Promise((r) => setTimeout(r, 10)); +d.destroy(); +await once(d, "close"); +console.log("closed d"); + +const seen = []; +const fin = Duplex.from(async function (source, { signal }) { + signal.addEventListener("abort", () => console.log("abort", seen.join(","))); + for await (const chunk of source) seen.push(String(chunk)); + console.log("done", signal.aborted); +}); +fin.on("finish", () => console.log("finish")); +fin.write("1"); +fin.write("2"); +fin.end(); +await once(fin, "close"); +console.log("closed fin"); + +const boom = new Error("boom"); +const e = Duplex.from(async function (source, { signal }) { + signal.addEventListener("abort", () => console.log("e abort", String(signal.reason), signal.reason === boom)); + await new Promise((r) => setTimeout(r, 50)); +}); +e.on("error", (err) => console.log("e error", err === boom)); +setTimeout(() => e.destroy(boom), 5); +await new Promise((r) => e.on("close", r)); +console.log("closed e"); + +const c = compose( + async function* (source, { signal }) { + for await (const chunk of source) yield `${chunk}!${signal.aborted}`; + }, + async function* (source, { signal }) { + for await (const chunk of source) yield chunk.toUpperCase(); + }, +); +Readable.from(["a", "b"]).pipe(c); +c.on("data", (x) => console.log("compose", String(x))); +c.on("end", () => console.log("compose end")); diff --git a/tests/corpus/7438-duplex-from-signal.ts b/tests/corpus/7438-duplex-from-signal.ts new file mode 100644 index 000000000..358bd3e68 --- /dev/null +++ b/tests/corpus/7438-duplex-from-signal.ts @@ -0,0 +1,26 @@ +// Typed Duplex.from bodies that declare the options argument: an async +// generator function and an async function. +import { Duplex } from "node:stream"; + +const d = Duplex.from(async function* (source: AsyncIterable, opts: { signal: AbortSignal }) { + opts.signal.addEventListener("abort", () => console.log("aborted", opts.signal.aborted)); + for await (const chunk of source) { + yield `<${chunk}>`; + } +}); +d.on("data", (c: string) => console.log("data", c)); +d.on("end", () => console.log("end")); +d.on("close", () => { + console.log("close"); + const w = Duplex.from(async function (source: AsyncIterable, { signal }: { signal: AbortSignal }): Promise { + let n = 0; + for await (const chunk of source) n += chunk.length; + console.log("total", n, signal.aborted); + }); + w.on("finish", () => console.log("finish")); + w.on("close", () => console.log("w close")); + w.write("abc"); + w.end("de"); +}); +d.write("x"); +d.end("y"); diff --git a/tests/corpus/7439-readable-read-string-chunks.ts b/tests/corpus/7439-readable-read-string-chunks.ts new file mode 100644 index 000000000..2fc5dbed6 --- /dev/null +++ b/tests/corpus/7439-readable-read-string-chunks.ts @@ -0,0 +1,17 @@ +// A typed read() answers the strings a stream holds: the first value a +// lazy Readable.from pull over a parsed array produces, and the decoded +// text once _read calls setEncoding(). +import { Readable } from "node:stream"; +const r = Readable.from(JSON.parse('["j1", "j2"]')); +const chunk = r.read(); +console.log("first", typeof chunk, chunk === null ? "null" : chunk.toString("hex")); +r.once("readable", () => console.log("later", JSON.stringify(r.read()))); +const e = new Readable({ + read() { + this.setEncoding("utf8"); + this.push(Buffer.from("ab")); + this.push(null); + }, +}); +const text = e.read(); +console.log("encoded", typeof text, text === null ? "null" : text.toString("hex")); diff --git a/tests/harness/corpus-costs/7433-readable-operator-signal.mjs.json b/tests/harness/corpus-costs/7433-readable-operator-signal.mjs.json new file mode 100644 index 000000000..3ef40f751 --- /dev/null +++ b/tests/harness/corpus-costs/7433-readable-operator-signal.mjs.json @@ -0,0 +1 @@ +{ "plain": 3885, "san": 5128 } diff --git a/tests/harness/corpus-costs/7434-readable-operator-values.ts.json b/tests/harness/corpus-costs/7434-readable-operator-values.ts.json new file mode 100644 index 000000000..fdbfd8cde --- /dev/null +++ b/tests/harness/corpus-costs/7434-readable-operator-values.ts.json @@ -0,0 +1 @@ +{ "plain": 2465, "san": 2933 } diff --git a/tests/harness/corpus-costs/7435-readable-flatmap-iterables.mjs.json b/tests/harness/corpus-costs/7435-readable-flatmap-iterables.mjs.json new file mode 100644 index 000000000..8c31cc905 --- /dev/null +++ b/tests/harness/corpus-costs/7435-readable-flatmap-iterables.mjs.json @@ -0,0 +1 @@ +{ "plain": 2265, "san": 2990 } diff --git a/tests/harness/corpus-costs/7436-readable-read-any-value.mjs.json b/tests/harness/corpus-costs/7436-readable-read-any-value.mjs.json new file mode 100644 index 000000000..c4a7cbc84 --- /dev/null +++ b/tests/harness/corpus-costs/7436-readable-read-any-value.mjs.json @@ -0,0 +1 @@ +{ "plain": 1675, "san": 2211 } diff --git a/tests/harness/corpus-costs/7437-duplex-from-signal.mjs.json b/tests/harness/corpus-costs/7437-duplex-from-signal.mjs.json new file mode 100644 index 000000000..dfbf933f4 --- /dev/null +++ b/tests/harness/corpus-costs/7437-duplex-from-signal.mjs.json @@ -0,0 +1 @@ +{ "plain": 2383, "san": 3146 } diff --git a/tests/harness/corpus-costs/7438-duplex-from-signal.ts.json b/tests/harness/corpus-costs/7438-duplex-from-signal.ts.json new file mode 100644 index 000000000..13b6f4240 --- /dev/null +++ b/tests/harness/corpus-costs/7438-duplex-from-signal.ts.json @@ -0,0 +1 @@ +{ "plain": 1777, "san": 2722 } diff --git a/tests/harness/corpus-costs/7439-readable-read-string-chunks.ts.json b/tests/harness/corpus-costs/7439-readable-read-string-chunks.ts.json new file mode 100644 index 000000000..a93aa2e95 --- /dev/null +++ b/tests/harness/corpus-costs/7439-readable-read-string-chunks.ts.json @@ -0,0 +1 @@ +{ "plain": 2578, "san": 3403 } diff --git a/tests/harness/self-hosting-native-toolchain.test.ts b/tests/harness/self-hosting-native-toolchain.test.ts index e74d9f60a..c26ef20c9 100644 --- a/tests/harness/self-hosting-native-toolchain.test.ts +++ b/tests/harness/self-hosting-native-toolchain.test.ts @@ -108,7 +108,7 @@ for (const backend of ["llvm"] as const) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" }, + runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" }, compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple }, macros: { executable: [], excluded: [], sanitizer: "external-toolchain-required" }, flavors: { diff --git a/tests/harness/stream-entry-kinds.test.ts b/tests/harness/stream-entry-kinds.test.ts index ec89c5914..b0aa52294 100644 --- a/tests/harness/stream-entry-kinds.test.ts +++ b/tests/harness/stream-entry-kinds.test.ts @@ -1,14 +1,9 @@ /* Stream chunk kinds Node cannot oracle — scriptc-only. * - * A typed read() answers `Buffer | null`. When the read itself makes the - * stream hold strings (the first string a lazy Readable.from pull over a - * dynamic array produces, or setEncoding() inside _read), Node returns - * the string, which the typed result cannot hold, so the read throws the - * encoding fence instead (caught here, so the program exits normally). - * These pins keep that refusal exact and, under - * SCRIPTC_SAN=1 (ASan and the RC audit, like every harness lane), prove - * no string reaches Buffer code. The Node-observable cases are corpus - * programs (stream-dynamic-entry-kinds.*). + * Under SCRIPTC_SAN=1 (ASan and the RC audit, like every harness lane) + * these pins prove no string reaches Buffer code. The Node-observable + * cases are corpus programs (stream-dynamic-entry-kinds.*, and + * 7439-readable-read-string-chunks for read() of strings). */ import { execFile } from "node:child_process"; import { createHash } from "node:crypto"; @@ -73,71 +68,7 @@ function stripAsanFiberWarning(s: string): string { ); } -const READ_FENCE_MESSAGE = - "read() on a stream with an encoding set is not supported yet (consume 'data' events, which deliver strings)"; - describe(`typed stream reads of string chunks (scriptc-only${sanitize ? ", sanitized" : ""})`, () => { - test("a typed read() whose lazy pull yields the first string throws the encoding fence", async () => { - const r = await compileAndRun( - "typed-read-json-string", - `import { Readable } from "node:stream"; -const r = Readable.from(JSON.parse('["j1", "j2"]')); -try { - const chunk = r.read(); - console.log("hex", chunk === null ? "null" : chunk.toString("hex")); -} catch (err) { - console.log("caught", (err as Error).message); -} -`, - ); - expect(r.exitCode).toBe(0); - expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`); - }); - - test("a typed read() whose _read calls setEncoding() throws the encoding fence", async () => { - const r = await compileAndRun( - "typed-read-encoding-in-read", - `import { Readable } from "node:stream"; -const r = new Readable({ - read() { - this.setEncoding("utf8"); - this.push(Buffer.from("ab")); - this.push(null); - }, -}); -try { - const chunk = r.read(); - console.log("hex", chunk === null ? "null" : chunk.toString("hex")); -} catch (err) { - console.log("caught", (err as Error).message); -} -`, - ); - expect(r.exitCode).toBe(0); - expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`); - }); - - test("the JavaScript lane's read() on a JSON source throws the same fence", async () => { - const r = await compileAndRun( - "js-read-json-string", - `'use strict'; -const { Readable } = require('node:stream'); -const r = Readable.from(JSON.parse('["j1"]')); -r.once('readable', () => { - try { - const chunk = r.read(); - console.log('got', chunk === null ? 'null' : chunk.length); - } catch (err) { - console.log('caught', err.message); - } -}); -`, - "cjs", - ); - expect(r.exitCode).toBe(0); - expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`); - }); - test("a dynamic source that mixes Buffers and strings yields each as it is, as Node does", async () => { const r = await compileAndRun( "mixed-dynamic-source",