From dfd59e18ee4d70cb2c5f432342c2629dd4dce728 Mon Sep 17 00:00:00 2001 From: Malte Ubl Date: Tue, 20 Oct 2026 06:49:37 -0700 Subject: [PATCH] Let NodeJS stream interface slots hold JavaScript stream objects Under the general stream representation, NodeJS.ReadableStream and NodeJS.WritableStream compile as the runtime Readable and Writable classes, so a JavaScript stream object (a constructor-function stream, a plain object with on/pipe, a stream from bundled npm code) passed through a TypeScript function typed with them took the checked unwrap and threw. A program that moves an untyped value into a runtime stream class slot, or passes an untyped pipe() destination or pipeline stage, now lowers once more with checked-dynamic stream slots: the two interfaces (and intersections over them) map to dyn. A native stream rides its typed capsule and keeps its identity; any other value is the JavaScript object itself. Member calls dispatch per value: stream.dynNative tests for a boxed native stream, which takes the ordinary class-method lowering; other values call their own member through dynInvoke. pipe() into a JavaScript destination follows Node's Readable.prototype.pipe listener protocol (stream.pipeDyn). for await and pipeline stages keep the not-supported error for values that are not native streams. The runtime code lives in scr_stream_dyn.c behind a new streamDyn feature, so programs without these libCalls link the same objects. Co-Authored-By: Claude Opus 5.5 --- .../builtin-apis/streams-and-readline.mdx | 5 +- .../src/backend/executable-features.ts | 1 + .../compiler/src/backend/llvm/lib-shared.ts | 1 + .../compiler/src/backend/llvm/lib-stream.ts | 18 + .../src/backend/native-link-info-core.ts | 3 + .../compiler/src/backend/native/contracts.ts | 3 + .../compiler/src/backend/native/executable.ts | 3 + .../compiler/src/backend/runtime-pack-core.ts | 2 + .../src/executable/executable-cache.ts | 1 + .../frontend/lowering/builtins/properties.ts | 3 + .../src/frontend/lowering/dyn-streams.ts | 314 ++++++++++++++++ .../src/frontend/lowering/general-streams.ts | 15 +- .../lowering/lower-async-iteration.ts | 4 +- .../src/frontend/lowering/lower-calls.ts | 4 + .../src/frontend/lowering/lower-stmts.ts | 6 + .../src/frontend/lowering/lower-stream.ts | 59 ++- .../compiler/src/frontend/lowering/lowerer.ts | 57 ++- packages/compiler/src/frontend/type-mapper.ts | 45 ++- packages/compiler/src/index.ts | 1 + packages/compiler/src/ir/builtin-effects.ts | 1 + .../compiler/src/ir/builtin-signatures.ts | 10 + packages/compiler/src/ir/runtime-features.ts | 8 + packages/compiler/src/ir/validate.ts | 15 + .../compiler/src/native/sanitizer.test.ts | 1 + .../runtime-pack-matrix.mjs | 2 + packages/runtime/src/scr_stream_dyn.c | 349 ++++++++++++++++++ 26 files changed, 916 insertions(+), 15 deletions(-) create mode 100644 packages/compiler/src/frontend/lowering/dyn-streams.ts create mode 100644 packages/runtime/src/scr_stream_dyn.c diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index 042fd1e50..bd49c89e4 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -1,7 +1,10 @@ - **Streams and readline:** Stream chunks are Buffers or strings. `NodeJS.ReadableStream` and `NodeJS.WritableStream`, and `stream.Readable` and `stream.Writable` under `@types/node`, accept every stream: when a program passes file, zlib, or `node:stream` streams through them, child-process pipes and `process.stdout` and `process.stderr` join those streams as `Readable` and `Writable` instances. `readable.pipe(process.stdout)` and `readable.pipe(process.stderr)` compile and, as in Node, do not end the process stream. - A `pipeline` stage that arrives as an untyped value must be a native stream at run time; other values throw a not-supported error. + When an untyped value reaches `NodeJS.ReadableStream` or `NodeJS.WritableStream`, those two interfaces hold any value, so JavaScript stream objects (constructor-function streams, plain objects with `on` and `pipe`) pass through TypeScript functions typed with them and come back to JavaScript as the same objects. + Member calls on such a value use the native stream when it is one and call the object's own members otherwise; `readable.pipe()` into a JavaScript destination follows Node's pipe protocol. + A `pipeline` stage and a `for await` source that arrive as untyped values must be native streams at run time; other values throw a not-supported error. + Native streams expose no `emit` to JavaScript, so a JavaScript stream whose own `pipe()` emits on its destination (the legacy `Stream` pipe) cannot pipe into one. `Readable.from` accepts arrays, Sets, Map and Set iterators, generators, async generators, and class iterables and async iterables whose values are strings or Buffers, and in JavaScript any iterable or async iterable of them; a stream that mixes strings and Buffers is unsupported. In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported. Object-mode streams of other values (including `Readable.from` over a `Map`, whose values are entry arrays) are unsupported. diff --git a/packages/compiler/src/backend/executable-features.ts b/packages/compiler/src/backend/executable-features.ts index eeb6c6900..74d5de278 100644 --- a/packages/compiler/src/backend/executable-features.ts +++ b/packages/compiler/src/backend/executable-features.ts @@ -25,6 +25,7 @@ export function executableLinkFeatures(mod: IrModule): NativeLinkFeatures { dc: features.dc, dynAsync: features.dynAsync, ...(features.dynTier ? { dynTier: true } : {}), + ...(features.streamDyn ? { streamDyn: true } : {}), events: features.processEvents, emitter: features.emitter, symbol: features.symbol, diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 4602a2bdd..6f296205c 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -728,6 +728,7 @@ export const LIB_FN_SYMS: Record = { "procStream.write": "scr_proc_stream_write", "stream.stdioWritable": "scr_stream_stdio_writable", "stream.unboxDyn": "scr_stream_unbox_dyn", + "stream.pipeDyn": "scr_stream_pipe_dyn", "process.kill": "scr_process_kill_named", "process.killNum": "scr_process_kill", // fs.promises (scr_bytes_io.c / scr_lib.c): the promise forms settle diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index b62b237df..21e5038cd 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -36,6 +36,24 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV host.emitPendingCheck(); return out; } + if (e.fn === "stream.dynNative") { + // Checked-dynamic stream slots: the boxed native stream (+1) or NULL. + if (e.type.kind !== "union") + throw new InternalCompilerError(`llvm emitter bug: ${e.fn} result is not a union`); + const def = host.unionsById.get(e.type.unionId); + const streamTag = def ? def.arms.findIndex((a) => a.kind === "object") : -1; + const nullTag = def ? def.arms.findIndex((a) => a.kind === "nullT") : -1; + if (streamTag < 0 || nullTag < 0) + throw new InternalCompilerError(`llvm emitter bug: ${e.fn} union lacks its arms`); + const value = host.emitExpr(e.args[0]!); + const side = host.emitExpr(e.args[1]!); + host.declare(`declare ptr @scr_stream_dyn_native(ptr, double)`); + const raw = B.tmp(); + B.line( + `${raw} = call ptr @scr_stream_dyn_native(ptr ${value.name}, double ${side.name}) ; +1 or NULL`, + ); + return host.wrapNullable(raw, raw, def!.arms[streamTag]!, streamTag, e.type, nullTag); + } if (e.fn === "stream.onData") { // The callback MOVES into the stream's listener registry; the // adapter is per callback shape — runtime-provided for the diff --git a/packages/compiler/src/backend/native-link-info-core.ts b/packages/compiler/src/backend/native-link-info-core.ts index 86146259d..df03d528a 100644 --- a/packages/compiler/src/backend/native-link-info-core.ts +++ b/packages/compiler/src/backend/native-link-info-core.ts @@ -40,6 +40,9 @@ export interface NativeLinkFeatures { /** Dynamic-tier jsval operations: the scr_js_* units. Absent (not false) * for programs without them, so their link features stay unchanged. */ dynTier?: boolean; + /** Checked-dynamic stream slots: the scr_stream_dyn.c unit. Absent (not + * false) for programs without them. */ + streamDyn?: boolean; events: boolean; emitter: boolean; symbol: boolean; diff --git a/packages/compiler/src/backend/native/contracts.ts b/packages/compiler/src/backend/native/contracts.ts index 33b13960e..3b700ea69 100644 --- a/packages/compiler/src/backend/native/contracts.ts +++ b/packages/compiler/src/backend/native/contracts.ts @@ -113,6 +113,9 @@ export interface CcOptions { * the IR): compiles the scr_js_* units (two-tier design I3). Programs * without them link no scr_js_ symbol. */ dynTier?: boolean; + /** Checked-dynamic stream slots (stream.dynNative / stream.pipeDyn): + * compiles scr_stream_dyn.c. */ + streamDyn?: boolean; /** The program constructs EventTarget/Event/CustomEvent values or * calls events.addAbortListener (the event.* libCalls), or uses the * native fetch AbortSignal built on the same engine: compiles diff --git a/packages/compiler/src/backend/native/executable.ts b/packages/compiler/src/backend/native/executable.ts index 5ebb4796e..d2616151b 100644 --- a/packages/compiler/src/backend/native/executable.ts +++ b/packages/compiler/src/backend/native/executable.ts @@ -240,6 +240,9 @@ const RUNTIME_UNITS: readonly RuntimeUnit[] = [ // stream unit. unit((u) => u.opts.zlib && u.opts.stream, "scr_zlib_stream.c"), unit((u) => u.opts.stream && u.opts.dynAsync, "scr_stream_ops.c"), + // Checked-dynamic stream slots test values and call JavaScript stream + // members through the dyn-invoke unit. + unit((u) => u.opts.streamDyn === true, "scr_stream_dyn.c"), // FileHandle streams reference both the stream unit and FileHandle. unit((u) => u.opts.stream && u.opts.fileHandle, "scr_fs_stream_fh.c"), // The readiness-poller backends (scr_platform.h): kqueue on macOS/BSD, diff --git a/packages/compiler/src/backend/runtime-pack-core.ts b/packages/compiler/src/backend/runtime-pack-core.ts index 6a660e00c..41fa1442a 100644 --- a/packages/compiler/src/backend/runtime-pack-core.ts +++ b/packages/compiler/src/backend/runtime-pack-core.ts @@ -83,6 +83,7 @@ export interface RuntimeFeatureSet extends NativeLinkFeatures { youngNursery: boolean; eventTarget: boolean; dynTier: boolean; + streamDyn: boolean; warningReplace: boolean; stdioExtended: boolean; nativeFetch: boolean; @@ -273,6 +274,7 @@ export function effectiveRuntimeFeatures( youngNursery: features.youngNursery === true, eventTarget: features.eventTarget === true, dynTier: features.dynTier === true, + streamDyn: features.streamDyn === true, warningReplace: features.warningReplace === true, stdioExtended: features.stdioExtended === true, nativeFetch, diff --git a/packages/compiler/src/executable/executable-cache.ts b/packages/compiler/src/executable/executable-cache.ts index 8c67c1006..7e3b427e5 100644 --- a/packages/compiler/src/executable/executable-cache.ts +++ b/packages/compiler/src/executable/executable-cache.ts @@ -65,6 +65,7 @@ export interface EarlyExecutableNativeFeatures { dc: boolean; dynAsync: boolean; dynTier?: boolean; + streamDyn?: boolean; events: boolean; emitter: boolean; symbol: boolean; diff --git a/packages/compiler/src/frontend/lowering/builtins/properties.ts b/packages/compiler/src/frontend/lowering/builtins/properties.ts index e60a14c0c..70512fed9 100644 --- a/packages/compiler/src/frontend/lowering/builtins/properties.ts +++ b/packages/compiler/src/frontend/lowering/builtins/properties.ts @@ -290,6 +290,9 @@ function generalChildStdioType( const className = name === "stdin" ? "%Writable" : "%Readable"; const mapped = lowerer.mapTypeOf(lowerer.typeOf(expr)); if (!mapped) return null; + // Checked-dynamic stream slots: the declared stream type maps to dyn + // (dyn-streams.ts); the read still produces the bridged stream. + if (lowerer.dynNodeStreams && mapped.kind === "dyn") return { kind: "object", className }; const arms = mapped.kind === "union" ? (lowerer.unions.get(mapped.unionId)?.arms ?? []) : [mapped]; return arms.some((arm) => arm.kind === "object" && arm.className === className) diff --git a/packages/compiler/src/frontend/lowering/dyn-streams.ts b/packages/compiler/src/frontend/lowering/dyn-streams.ts new file mode 100644 index 000000000..9694e3820 --- /dev/null +++ b/packages/compiler/src/frontend/lowering/dyn-streams.ts @@ -0,0 +1,314 @@ +/* Checked-dynamic stream slots (TypeMapperCtx.dynNodeStreams). + * + * Under the general stream representation (general-streams.ts), + * NodeJS.ReadableStream and NodeJS.WritableStream map to the runtime stream + * classes. JavaScript code also hands such slots its own stream objects: + * constructor-function streams over EventEmitter or the legacy Stream, plain + * objects implementing on/pipe/write/end, streams of bundled npm code. A + * program that moves an untyped value into a stream slot therefore lowers + * again with checked-dynamic slots: those two interfaces map to dyn, so a + * slot holds either a boxed native stream (its typed capsule, which unboxes + * to the same stream) or the JavaScript object itself, and a value returned + * to JavaScript is the object that came in. + * + * A member call on such a slot dispatches per value at run time. A native + * stream unboxes and takes the ordinary class-method lowering (the same + * stream surface a typed stream has); any other value calls its own member + * with JavaScript's dynamic semantics (dynInvoke). Programs without such a + * flow never take this pass, so their IR is unchanged. */ +import type { IrExpr, IrStmt, IrType } from "../../ir/ir.js"; +import { dynUndefinedExpr, numLit, strLit, varRef } from "../../ir/build.js"; +import { DYN } from "../../ir/ir.js"; +import * as ts from "../ts7/adapter.js"; +import { locOf } from "../program.js"; +import { isDeclaredInAmbientNamespace } from "../type-mapper.js"; +import { resultIsDiscarded } from "./call-position.js"; +import { lowerObjectMethodCall } from "./lower-calls.js"; +import type { Lowerer } from "./lowerer.js"; + +/** The side a NodeJS stream interface names: "r" for NodeJS.ReadableStream, + * "w" for NodeJS.WritableStream (through null/undefined and intersections), + * or null for any other type. */ +export function nodeStreamInterfaceSide(lowerer: Lowerer, type: ts.Type): "r" | "w" | null { + const parts = (t: ts.Type): ts.Type[] => + (t.flags & ts.TypeFlags.Intersection) !== 0 ? ts.constituentTypes(t).flatMap(parts) : [t]; + for (const part of parts(lowerer.checker.getNonNullableType(type))) { + const psym = part.getSymbol() ?? part.getAliasSymbol(); + if (psym?.name !== "ReadableStream" && psym?.name !== "WritableStream") continue; + const declared = psym.declarations.some( + (d) => + ts.isInterfaceDeclaration(d) && + lowerer.isStdlibFile(d.getSourceFile()) && + isDeclaredInAmbientNamespace(d, "NodeJS"), + ); + if (declared) return psym.name === "ReadableStream" ? "r" : "w"; + } + return null; +} + +/** The runtime class a native value of the side unboxes to. */ +export function nativeStreamClass(side: "r" | "w"): IrType { + return { kind: "object", className: side === "r" ? "%Readable" : "%Writable" }; +} + +/** The native stream of a checked-dynamic value (+1, typed as the side's + * class) or null: the stream.dynNative libCall, which tests the value's + * typed capsule without throwing. */ +function dynNativeStream(lowerer: Lowerer, value: IrExpr, side: "r" | "w"): IrExpr { + const native = nativeStreamClass(side); + return { + kind: "libCall", + fn: "stream.dynNative", + args: [value, numLit(side === "r" ? 1 : 2, value.loc)], + type: { kind: "union", unionId: lowerer.unions.intern([native, { kind: "nullT" }]) }, + loc: value.loc, + }; +} + +/** Runs `lower` with `node` lowering to `value` and its member calls taking + * `type` as the receiver representation. */ +function withStreamReceiver( + lowerer: Lowerer, + node: ts.Expression, + value: IrExpr, + type: IrType, + lower: () => T, +): T { + const previous = lowerer.receiverTypeOverrides.get(node); + lowerer.receiverTypeOverrides.set(node, type); + try { + return lowerer.withExpressionOverride(node, value, lower); + } finally { + if (previous) lowerer.receiverTypeOverrides.set(node, previous); + else lowerer.receiverTypeOverrides.delete(node); + } +} + +/** `stream.member(args)` on a NodeJS.ReadableStream / NodeJS.WritableStream + * receiver under checked-dynamic stream slots, or null when the call is not + * one. */ +export function lowerDynStreamMemberCall( + lowerer: Lowerer, + call: ts.CallExpression, + access: ts.PropertyAccessExpression, +): IrExpr | null { + if (!lowerer.dynNodeStreams) return null; + // The native arm re-enters the call lowering with the receiver bound. + if (lowerer.chainRecvByNode.has(access.expression)) return null; + if (access.questionDotToken || call.questionDotToken || !ts.isIdentifier(access.name)) + return null; + if (call.arguments.nodes.some(ts.isSpreadElement)) return null; + const side = nodeStreamInterfaceSide(lowerer, lowerer.typeOf(access.expression)); + if (side === null) return null; + const loc = locOf(call); + const receiverValue = lowerer.lowerExpr(access.expression); + const receiver = lowerer.declareHiddenLocal("%streamReceiver", receiverValue.type); + const receiverRef = varRef(receiver.id, receiver.type, locOf(access.expression)); + const stmts: IrStmt[] = [{ kind: "varDecl", localId: receiver.id, init: receiverValue, loc }]; + const seq = (result: IrExpr): IrExpr => ({ + kind: "seqExpr", + stmts, + result, + type: result.type, + loc, + }); + // A typed stream whose declared type is one of the interfaces (a child + // stdio read): the class-method path on its own representation. + if (receiverValue.type.kind !== "dyn") { + const typed = + receiverValue.type.kind === "object" + ? withStreamReceiver(lowerer, access.expression, receiverRef, receiverValue.type, () => + lowerObjectMethodCall(lowerer, call, access), + ) + : lowerer.withExpressionOverride(access.expression, receiverRef, () => + lowerer.lowerCall(call), + ); + if (typed === null) lowerer.noLowering(`${access.name.text} on this stream`, call); + return seq(typed); + } + const nativeType = nativeStreamClass(side); + const nullable = dynNativeStream(lowerer, receiverRef, side); + const nativeOrNull = lowerer.declareHiddenLocal("%streamNative", nullable.type); + stmts.push({ kind: "varDecl", localId: nativeOrNull.id, init: nullable, loc }); + const unionId = nullable.type.kind === "union" ? nullable.type.unionId : ""; + const tag = lowerer.armTag(unionId, nativeType); + const native = lowerer.declareHiddenLocal("%streamNativeValue", nativeType); + const nativeRef = varRef(native.id, nativeType, locOf(access.expression)); + const nativeDecl: IrStmt = { + kind: "varDecl", + localId: native.id, + init: { + kind: "unionNarrow", + unionId, + tag, + value: varRef(nativeOrNull.id, nullable.type, loc), + type: nativeType, + loc, + }, + loc, + }; + const typed = withStreamReceiver(lowerer, access.expression, nativeRef, nativeType, () => + lowerObjectMethodCall(lowerer, call, access), + ); + if (typed === null) + lowerer.noLowering( + `${side === "r" ? "NodeJS.ReadableStream" : "NodeJS.WritableStream"}.${access.name.text}`, + call, + ); + // Any other value calls its own member, with the arguments as untyped + // values (typed callbacks box behind checked thunks). + const invoked: IrExpr = { + kind: "dynInvoke", + recv: receiverRef, + method: access.name.text, + calleeName: access.getText(), + args: call.arguments.nodes.map((arg) => lowerer.lowerExprExpecting(arg, DYN)), + type: DYN, + loc, + }; + const cond: IrExpr = { + kind: "unionIsTag", + unionId, + tag, + negated: false, + value: varRef(nativeOrNull.id, nullable.type, loc), + type: { kind: "bool" }, + loc, + }; + const resultType = resultIsDiscarded(call) ? null : lowerer.irTypeOf(call); + if (resultType === null || resultType.kind === "void") { + stmts.push({ + kind: "if", + cond, + then: [nativeDecl, { kind: "exprStmt", expr: typed, loc }], + else_: [{ kind: "exprStmt", expr: invoked, loc }], + loc, + }); + return seq(dynUndefinedExpr(loc)); + } + const result = lowerer.declareHiddenLocal("%streamResult", resultType); + result.mutable = true; + stmts.push( + { kind: "varDecl", localId: result.id, init: null, loc }, + { + kind: "if", + cond, + then: [ + nativeDecl, + { + kind: "assign", + localId: result.id, + value: lowerer.coerceInto(call, typed, resultType), + loc, + }, + ], + else_: [ + { + kind: "assign", + localId: result.id, + value: lowerer.coerceInto(call, invoked, resultType), + loc, + }, + ], + loc, + }, + ); + return seq(varRef(result.id, resultType, loc)); +} + +/** `for await (const chunk of stream)` over a checked-dynamic + * NodeJS.ReadableStream: a native stream iterates through the Readable + * lowering; any other value throws the not-supported Error (async + * iteration of JavaScript stream objects has no lowering). */ +export function lowerForAwaitDynStream( + lowerer: Lowerer, + stmt: ts.ForOfStatement, + lowerLoop: () => IrStmt | null, +): IrStmt | null { + if (!lowerer.dynNodeStreams || lowerer.chainRecvByNode.has(stmt.expression)) return null; + if (nodeStreamInterfaceSide(lowerer, lowerer.typeOf(stmt.expression)) !== "r") return null; + const loc = locOf(stmt.expression); + const value = lowerer.lowerExpr(stmt.expression); + const native: IrType = + value.type.kind === "dyn" ? { kind: "object", className: "%Duplex" } : value.type; + const local = lowerer.declareHiddenLocal("%streamIterated", native); + const init: IrExpr = + value.type.kind === "dyn" + ? { + kind: "libCall", + fn: "stream.unboxDyn", + args: [ + value, + strLit( + "for await over stream objects other than native streams is not supported yet", + loc, + ), + ], + type: native, + loc, + } + : value; + const loop = withStreamReceiver( + lowerer, + stmt.expression, + varRef(local.id, native, loc), + native, + lowerLoop, + ); + if (loop === null) return null; + return { + kind: "block", + body: [{ kind: "varDecl", localId: local.id, init, loc }, loop], + loc: locOf(stmt), + }; +} + +/** A statement that runs `onNative` with the native stream a + * checked-dynamic value boxes (typed as the side's class), or `otherwise` + * for any other value. */ +export function dynNativeBranch( + lowerer: Lowerer, + value: IrExpr, + side: "r" | "w", + onNative: (native: IrExpr) => IrStmt, + otherwise: IrStmt, +): IrStmt { + const loc = value.loc; + const nativeType = nativeStreamClass(side); + const nullable = dynNativeStream(lowerer, value, side); + const nativeOrNull = lowerer.declareHiddenLocal("%streamNative", nullable.type); + const unionId = nullable.type.kind === "union" ? nullable.type.unionId : ""; + const tag = lowerer.armTag(unionId, nativeType); + const native = lowerer.declareHiddenLocal("%streamNativeValue", nativeType); + const nullableRef = varRef(nativeOrNull.id, nullable.type, loc); + return { + kind: "block", + body: [ + { kind: "varDecl", localId: nativeOrNull.id, init: nullable, loc }, + { + kind: "if", + cond: { + kind: "unionIsTag", + unionId, + tag, + negated: false, + value: nullableRef, + type: { kind: "bool" }, + loc, + }, + then: [ + { + kind: "varDecl", + localId: native.id, + init: { kind: "unionNarrow", unionId, tag, value: nullableRef, type: nativeType, loc }, + loc, + }, + onNative(varRef(native.id, nativeType, loc)), + ], + else_: [otherwise], + loc, + }, + ], + loc, + }; +} diff --git a/packages/compiler/src/frontend/lowering/general-streams.ts b/packages/compiler/src/frontend/lowering/general-streams.ts index 0113a9958..2cb6a4b6c 100644 --- a/packages/compiler/src/frontend/lowering/general-streams.ts +++ b/packages/compiler/src/frontend/lowering/general-streams.ts @@ -88,9 +88,11 @@ export function stdioIntoGeneralStream( ): IrExpr | null { if (!lowerer.generalNodeStreams || expr.type.kind !== "procStream") return null; const writable: IrType = { kind: "object", className: "%Writable" }; + // Checked-dynamic stream slots (dyn-streams.ts) hold the Writable boxed. const accepts = (expected.kind === "object" && expected.className === "%Writable") || - (expected.kind === "union" && lowerer.armTag(expected.unionId, writable) >= 0); + (expected.kind === "union" && lowerer.armTag(expected.unionId, writable) >= 0) || + (expected.kind === "dyn" && lowerer.dynNodeStreams); if (!accepts) return null; return { kind: "libCall", @@ -112,3 +114,14 @@ export function requestGeneralStreamSurface(lowerer: Lowerer, receiver: ts.Expre if (isGeneralNodeStreamInterface(type, lowerer.typeCtx)) lowerer.requiresGeneralNodeStreams = true; } + +/** Under the general representation, an untyped value reaching a runtime + * stream class slot (the checked unwrap) or an untyped pipe/pipeline + * stage: the program lowers again with checked-dynamic stream slots + * (dyn-streams.ts), where NodeJS.ReadableStream / NodeJS.WritableStream + * hold JavaScript stream objects too. */ +export function noteUntypedStreamFlow(lowerer: Lowerer, expected: IrType | null): void { + if (!lowerer.generalNodeStreams || lowerer.dynNodeStreams) return; + if (expected === null || containsPart(lowerer, expected, (p) => isGeneralStream(lowerer, p))) + lowerer.requiresDynNodeStreams = true; +} diff --git a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts index f095c5b9c..0e48f2040 100644 --- a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts +++ b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts @@ -137,7 +137,9 @@ function literalBooleanOption( function nodeReadablePlan(lowerer: Lowerer, expression: ts.Expression): NodeReadablePlan | null { const direct = - lowerer.stageSourceType(expression) ?? lowerer.mapTypeOf(lowerer.typeOf(expression)); + lowerer.receiverTypeOverrides.get(expression) ?? + lowerer.stageSourceType(expression) ?? + lowerer.mapTypeOf(lowerer.typeOf(expression)); if (direct?.kind === "object") { const sides = streamSidesOf(lowerer, lowerer.classes.get(direct.className)); if (sides === "r" || sides === "rw") { diff --git a/packages/compiler/src/frontend/lowering/lower-calls.ts b/packages/compiler/src/frontend/lowering/lower-calls.ts index ae864c2cb..9be693ef5 100644 --- a/packages/compiler/src/frontend/lowering/lower-calls.ts +++ b/packages/compiler/src/frontend/lowering/lower-calls.ts @@ -1,4 +1,5 @@ import { guardNullProto, nullProtoRecordShape } from "./null-prototype.js"; +import { lowerDynStreamMemberCall } from "./dyn-streams.js"; import { refuseModuleMemberThis } from "./module-member-this.js"; import { lowerUnionEquality, tagEqualityMayMissAlias } from "./strict-equality.js"; import { lowerHashStreamCall } from "./builtins/hash-stream.js"; @@ -1397,6 +1398,8 @@ export function lowerCall(lowerer: Lowerer, expr: ts.CallExpression): IrExpr { } if (ts.isPropertyAccessExpression(expr.expression)) { const intrinsic = + // Checked-dynamic stream slots dispatch per value (dyn-streams.ts). + lowerDynStreamMemberCall(lowerer, expr, expr.expression) ?? fenceNodeModuleMutationCall(lowerer, expr, expr.expression) ?? // Builtin namespace imports first (`fs.readFileSync(...)` where fs // is `import * as fs from "node:fs"`): the same tables and fences @@ -9289,6 +9292,7 @@ export function lowerObjectMethodCall( ): IrExpr | null { if (lowerer.chainBlocked(access, call)) return null; let mappedReceiver = + lowerer.receiverTypeOverrides.get(access.expression) ?? structuralReadType(lowerer, access.expression) ?? lowerer.mapTypeOf(lowerer.typeOf(access.expression)); // EventEmitter's static side (the class, a value of it, or the events diff --git a/packages/compiler/src/frontend/lowering/lower-stmts.ts b/packages/compiler/src/frontend/lowering/lower-stmts.ts index afc329f40..03150f0a3 100644 --- a/packages/compiler/src/frontend/lowering/lower-stmts.ts +++ b/packages/compiler/src/frontend/lowering/lower-stmts.ts @@ -1,5 +1,6 @@ import { hostsAll } from "./expressions/hosted-narrowing.js"; import { requestGeneralStreamSurface } from "./general-streams.js"; +import { lowerForAwaitDynStream } from "./dyn-streams.js"; import { matchAllRowsReadByElements, storedMatchAllRowsReadByElements, @@ -8915,6 +8916,11 @@ export function lowerForOf(lowerer: Lowerer, stmt: ts.ForOfStatement): IrStmt { ) { return lowerForAwaitStdin(lowerer, stmt); } + // A checked-dynamic NodeJS.ReadableStream iterates its native stream. + const dynStream = lowerForAwaitDynStream(lowerer, stmt, () => + lowerForAwaitBuiltin(lowerer, stmt), + ); + if (dynStream) return dynStream; const builtin = lowerForAwaitBuiltin(lowerer, stmt); if (builtin) return builtin; // A NodeJS.ReadableStream iterates as the general Readable. diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 51c86fe39..53e2927d5 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -49,7 +49,8 @@ import { sideEffectFreeOptionValue } from "./surfaces.js"; import { lowerDynObjectLiteral } from "./expressions/object-literals.js"; import { dynUndefinedExpr, varRef } from "../../ir/build.js"; import { genResultRecord } from "../type-mapper.js"; -import { requestGeneralStreamSurface } from "./general-streams.js"; +import { noteUntypedStreamFlow, requestGeneralStreamSurface } from "./general-streams.js"; +import { dynNativeBranch } from "./dyn-streams.js"; import { extractIteratorValue, genStep, genValue, stepDrivable } from "./lower-generators.js"; import { BOOL, @@ -2049,13 +2050,16 @@ function lowerStreamArg( // a boxed native stream unwraps at run time; other values (iterables, // functions) throw the not-supported Error there. if (v.type.kind === "dyn" && what === "pipeline") { + noteUntypedStreamFlow(lowerer, null); return { kind: "libCall", fn: "stream.unboxDyn", args: [ v, strLit( - "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet", + lowerer.dynNodeStreams + ? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet" + : "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet", locOf(node), ), ], @@ -3381,6 +3385,7 @@ function pipeIntoDyn( target: IrExpr, ): IrExpr { const loc = locOf(call); + noteUntypedStreamFlow(lowerer, null); const receiverValue = lowerer.lowerExpr(access.expression); const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type); const destination = lowerer.declareHiddenLocal("%pipeDestination", target.type); @@ -3390,6 +3395,56 @@ function pipeIntoDyn( ]; const end = pipeEndOption(lowerer, call.arguments.nodes[1], loc); const duplex: IrType = { kind: "object", className: "%Duplex" }; + // Checked-dynamic stream slots (dyn-streams.ts): a JavaScript stream + // object is a destination too, piped with Node's listener protocol. + if (lowerer.dynNodeStreams) { + const sourceRef: IrExpr = { kind: "varRef", localId: source.id, type: receiverValue.type, loc }; + const destinationRef: IrExpr = { + kind: "varRef", + localId: destination.id, + type: target.type, + loc, + }; + const ending = lowerer.declareHiddenLocal("%pipeEnd", BOOL); + stmts.push({ kind: "varDecl", localId: ending.id, init: end, loc }); + const endRef: IrExpr = { kind: "varRef", localId: ending.id, type: BOOL, loc }; + stmts.push( + dynNativeBranch( + lowerer, + destinationRef, + "w", + (native) => ({ + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "readable.pipe", + args: [sourceRef, native, endRef], + type: native.type, + loc, + }, + loc, + }), + { + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "stream.pipeDyn", + args: [sourceRef, lowerer.coerceToExpected(sourceRef, DYN), destinationRef, endRef], + type: VOID, + loc, + }, + loc, + }, + ), + ); + return { + kind: "seqExpr", + stmts, + result: destinationRef, + type: target.type, + loc, + }; + } stmts.push({ kind: "exprStmt", expr: { diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts index c9a0c6435..f31ed99c8 100644 --- a/packages/compiler/src/frontend/lowering/lowerer.ts +++ b/packages/compiler/src/frontend/lowering/lowerer.ts @@ -1,4 +1,8 @@ -import { needsGeneralNodeStreams, stdioIntoGeneralStream } from "./general-streams.js"; +import { + needsGeneralNodeStreams, + noteUntypedStreamFlow, + stdioIntoGeneralStream, +} from "./general-streams.js"; import { extendAssertedRecordMembers } from "./assertion-members.js"; import { jsDynamicDeclarations } from "./js-dynamic-objects.js"; import { @@ -882,6 +886,11 @@ export interface LowererMode { * generalNodeStreams). lowerToIr turns it on only after a pass without * it saw a stream flow that needs it (requiresGeneralNodeStreams). */ generalNodeStreams?: boolean; + /** With generalNodeStreams, the checked-dynamic stream slots + * (TypeMapperCtx.dynNodeStreams). lowerToIr turns it on only after a + * pass saw an untyped value flow into a stream slot + * (requiresDynNodeStreams). */ + dynNodeStreams?: boolean; /** LoadResult.inlineRequireCycleCalls. */ inlineRequireCycleCalls?: ReadonlySet | undefined; /** LowerOptions.libraryCallbacks (see there). */ @@ -1047,7 +1056,11 @@ export function lowerToIr( // a general stream into a stdio-represented stream slot lowers again // with it. Every other program keeps the single pass above unchanged. const generalNodeStreams = reachableEmit.requiresGeneralNodeStreams; - if (generalNodeStreams) { + // Checked-dynamic stream slots (dyn-streams.ts) are a further per-program + // decision under the general representation: only a program that moves an + // untyped value into a stream slot lowers again with them. + let dynNodeStreams = reachableEmit.requiresDynNodeStreams; + const streamPass = (dyn: boolean): Lowerer => { const general = new Lowerer(program, entry, moduleOrder, { frontendServices: options.frontendServices, projectCheckJs: options.projectCheckJs ?? false, @@ -1063,6 +1076,7 @@ export function lowerToIr( tierMode, ffiBindingSymbols: ffiValidation.symbolsByName, generalNodeStreams: true, + ...(dyn ? { dynNodeStreams: true } : {}), }); for (const d of extraModuleCycleDiags) general.pushDiag(d); for (const d of ffiValidation.diagnostics) general.pushDiag(d); @@ -1070,7 +1084,15 @@ export function lowerToIr( reachable = generalEmitted.reachable; result = generalEmitted.result; resultLowerer = general; - timing("general-streams-relower", { reachable: reachable.size }); + timing(dyn ? "dyn-streams-relower" : "general-streams-relower", { reachable: reachable.size }); + return general; + }; + if (generalNodeStreams) { + const general = streamPass(dynNodeStreams); + if (!dynNodeStreams && general.requiresDynNodeStreams) { + dynNodeStreams = true; + streamPass(true); + } } // A generic class-rest support decision can depend on record metadata // whose historical owner is discovered only while retained bodies lower. @@ -1084,6 +1106,7 @@ export function lowerToIr( frontendServices: options.frontendServices, projectCheckJs: options.projectCheckJs ?? false, ...(generalNodeStreams ? { generalNodeStreams: true } : {}), + ...(dynNodeStreams ? { dynNodeStreams: true } : {}), reachable, targetPlatform, startupCrash, @@ -1140,11 +1163,12 @@ export function lowerToIr( }; else if (contracts.count > 0) result = { ...result, contracts }; if (options.coverage !== true) return result; - const remainderPass = (general: boolean): Lowerer => + const remainderPass = (general: boolean, dyn = dynNodeStreams): Lowerer => new Lowerer(program, entry, moduleOrder, { frontendServices: options.frontendServices, projectCheckJs: options.projectCheckJs ?? false, ...(general ? { generalNodeStreams: true } : {}), + ...(general && dyn ? { dynNodeStreams: true } : {}), reachable, remainder: true, alreadyFlushed: resultLowerer.flushedSymbols, @@ -1165,7 +1189,11 @@ export function lowerToIr( // stream flow there that needs the general representation lowers the // remainder again with it, as reachable emit would. if (!generalNodeStreams && remainder.requiresGeneralNodeStreams) { - remainder = remainderPass(true); + remainder = remainderPass(true, remainder.requiresDynNodeStreams); + rem = remainder.run(); + } + if (!dynNodeStreams && remainder.generalNodeStreams && remainder.requiresDynNodeStreams) { + remainder = remainderPass(true, true); rem = remainder.run(); } timing("coverage-remainder"); @@ -2548,6 +2576,15 @@ export class Lowerer { * a stdio handle reached an 'unknown' slot. lowerToIr then lowers the * program again under the general stream representation. */ requiresGeneralNodeStreams = false; + /** An untyped value reached a stream slot (a checked unwrap into a + * runtime stream class, or an untyped pipe/pipeline stage). lowerToIr + * then lowers the program again with checked-dynamic stream slots + * (dyn-streams.ts). */ + requiresDynNodeStreams = false; + /** Receiver representations forced by the per-value stream dispatch + * (dyn-streams.ts): the native arm lowers a member call on the unboxed + * stream through the ordinary class-method path. */ + readonly receiverTypeOverrides = new Map(); /** Helpers that snapshot shape declaration order into their bodies. * Reachability lowers inits before the declarations they discover, so * these rebuild after the worklist restores historical shape metadata. */ @@ -2996,6 +3033,7 @@ export class Lowerer { * committed refusals in "auto" and keeps them refusals in "static-only". */ readonly tierMode: TierMode; readonly generalNodeStreams: boolean; + readonly dynNodeStreams: boolean; /** Functions finishModule placed in the dynamic tier (ir/tiers.ts). */ dynamicFunctions: DynamicFunction[] = []; /** Their IR, for coverage's dynamic statement count. */ @@ -3125,6 +3163,7 @@ export class Lowerer { this.targetPlatform = mode.targetPlatform ?? process.platform; this.tierMode = mode.tierMode ?? "auto"; this.generalNodeStreams = mode.generalNodeStreams ?? false; + this.dynNodeStreams = this.generalNodeStreams && (mode.dynNodeStreams ?? false); this.workerTargets = mode.workerTargets ?? []; this.workerTargets.forEach((sf, id) => { this.workerTargetIdByPath.set(tsgoPath(resolve(sf.fileName)), id); @@ -3222,6 +3261,7 @@ export class Lowerer { this.externalTypeSpecifiersByFile.has(tsgoPath(resolve(sf.fileName))), typeMemo: this.typeMemo, ...(this.generalNodeStreams ? { generalNodeStreams: true } : {}), + ...(this.dynNodeStreams ? { dynNodeStreams: true } : {}), canMemoizeType: () => this.typeParamBindings === null && this.typeParamTsBindings === null && @@ -8506,8 +8546,12 @@ export class Lowerer { * validator ICEs on them as the usual backstop. */ requireExactShape(node: ts.Node, actual: IrType, expected: IrType): void { if (typeEquals(actual, expected)) return; - if (!this.generalNodeStreams && needsGeneralNodeStreams(this, actual, expected)) + if (!this.generalNodeStreams && needsGeneralNodeStreams(this, actual, expected)) { this.requiresGeneralNodeStreams = true; + // An untyped value into a stream slot also needs the checked-dynamic + // slots (dyn-streams.ts). + if (actual.kind === "dyn") this.requiresDynNodeStreams = true; + } // dyn mismatches first. A dyn ('unknown') value flowing into a typed // slot needs a CHECKED cast — the hint points at `as `; tsc // usually rejects this before we do, so the fence is mostly defensive. @@ -8888,6 +8932,7 @@ export class Lowerer { (id) => this.unions.get(id), ); const handleOk = DYN_HANDLE_KINDS.has(expected.kind); + noteUntypedStreamFlow(this, expected); if ( canDynCheckTo( expected, diff --git a/packages/compiler/src/frontend/type-mapper.ts b/packages/compiler/src/frontend/type-mapper.ts index 7acb539d1..3e6fb3be2 100644 --- a/packages/compiler/src/frontend/type-mapper.ts +++ b/packages/compiler/src/frontend/type-mapper.ts @@ -1178,6 +1178,12 @@ export interface TypeMapperCtx { * from every stream producer share their slots. Off (the default), the * stdio handles keep those spellings. */ generalNodeStreams?: boolean; + /** With generalNodeStreams, the program's checked-dynamic stream slots + * (lowerToIr decides it per program, dyn-streams.ts): NodeJS.ReadableStream + * and NodeJS.WritableStream map to the checked-dynamic representation, so + * a slot holds a native stream (a boxed runtime stream, identity kept) or + * a JavaScript stream object. Members dispatch per value at run time. */ + dynNodeStreams?: boolean; /** Successful context-free mappings, keyed by the * TypeScript server's stable type id. A generated facade can reference the * same large record/function type thousands of times; remapping its full @@ -3033,6 +3039,27 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null { // Under the program's general stream representation (generalNodeStreams) // NodeJS.WritableStream is the runtime Writable instead: the process // streams convert into it where they flow (process.stdioWritable). + // Checked-dynamic stream slots: an intersection over a NodeJS stream + // interface (`NodeJS.ReadableStream & { destroy?(): void }`) holds the + // same values as the interface. + if ( + ctx.dynNodeStreams === true && + (type.flags & ts.TypeFlags.Intersection) !== 0 && + ts.constituentTypes(type).some((part) => { + const sym = part.getSymbol(); + return ( + (sym?.name === "ReadableStream" || sym?.name === "WritableStream") && + sym.declarations.some( + (d) => + ts.isInterfaceDeclaration(d) && + ctx.isStdlibFile(d.getSourceFile()) && + isDeclaredInAmbientNamespace(d, "NodeJS"), + ) + ); + }) + ) { + return DYN; + } const nodeWritableInterface = psym?.name === "WritableStream" && psym.declarations.some( @@ -3042,7 +3069,7 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null { isDeclaredInAmbientNamespace(d, "NodeJS"), ); if (nodeWritableInterface && ctx.generalNodeStreams === true) { - return { kind: "object", className: "%Writable" }; + return ctx.dynNodeStreams === true ? DYN : { kind: "object", className: "%Writable" }; } if ( (psym?.name === "WriteStream" && @@ -3168,8 +3195,13 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null { )) ) { // The general representation: the runtime Readable, which child - // stdio reads produce through the child stream bridge. - if (ctx.generalNodeStreams === true) return { kind: "object", className: "%Readable" }; + // stdio reads produce through the child stream bridge. With + // checked-dynamic stream slots, NodeJS.ReadableStream holds any + // stream value (dyn-streams.ts); stream.Readable keeps the class. + if (ctx.generalNodeStreams === true) { + if (ctx.dynNodeStreams === true && psym.name === "ReadableStream") return DYN; + return { kind: "object", className: "%Readable" }; + } return { kind: "childStream" }; } // dgram.Socket: the same ambiguous "Socket" name disambiguated by its @@ -4195,7 +4227,12 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null { (ctx.generalNodeStreams === true && arm.kind === "object" && ((expected === "childStream" && arm.className === "%Readable") || - (expected === "childWriter" && arm.className === "%Writable"))); + (expected === "childWriter" && arm.className === "%Writable"))) || + // Checked-dynamic stream slots: `NodeJS.ReadableStream | null` + // collapses to dyn (the reads still produce the bridged stream). + (ctx.dynNodeStreams === true && + arm.kind === "dyn" && + (expected === "childStream" || expected === "childWriter")); return ( arms !== undefined && arms.some(isExpected) && diff --git a/packages/compiler/src/index.ts b/packages/compiler/src/index.ts index 7d8d88889..12b84654b 100644 --- a/packages/compiler/src/index.ts +++ b/packages/compiler/src/index.ts @@ -537,6 +537,7 @@ async function compileExecutableNative( dc: features.dc, dynAsync: features.dynAsync, ...(features.dynTier ? { dynTier: true } : {}), + ...(features.streamDyn ? { streamDyn: true } : {}), events: features.events, emitter: features.emitter, symbol: features.symbol, diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index 1c390bbe4..fa171b8a1 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -77,6 +77,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "procStream.write", "child.stdioStream", "stream.unboxDyn", + "stream.pipeDyn", "weakMap.new", "weakSet.new", "dyn.fromEntries", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 597076cb1..2d70a7254 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -2571,6 +2571,16 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ argTypes: [DYN, STRING], result: { kind: "object", className: "%Duplex" }, }, + /** Checked-dynamic stream slots (dyn-streams.ts): the native stream a + * checked-dynamic value boxes (+1) — a runtime stream with the readable + * (1) or writable (2) side — or null for any other value. Never throws; + * the result union is `%Readable | null` or `%Writable | null`. */ + "stream.dynNative": { argTypes: [DYN, F64], result: VOID }, + /** readable.pipe(destination, end) into a checked-dynamic destination + * that is not a native stream: [source, the source boxed, destination, + * end]. Node's pipe listener protocol over the destination's members + * (scr_stream_dyn.c); throws what those members throw. */ + "stream.pipeDyn": { argTypes: [null, DYN, DYN, BOOL], result: VOID }, "stream.stdioWritable": { argTypes: [PROCSTREAM_T], result: { kind: "object", className: "%Writable" }, diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 99ba76f79..9b45fad1a 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -163,6 +163,8 @@ const STREAM_LIB_FNS: ReadonlySet = new Set([ "stream.destroyErr", "stream.errored", "stream.prop", + "stream.dynNative", + "stream.pipeDyn", "stream.stdioWritable", "stream.unboxDyn", "stream.workerStdio", @@ -199,6 +201,8 @@ const READABLE_DYN_ASYNC_PREFIXES: readonly string[] = ["readable.fromGen", "rea /** Entry points that link the dyn-invoke unit. */ const DYN_INVOKE_LIB_FNS: ReadonlySet = new Set([ + "stream.dynNative", + "stream.pipeDyn", "dyn.arrayProtoCall", "dyn.arrayPrototype", "dyn.builtinMethod", @@ -502,6 +506,9 @@ const RUNTIME_FEATURE_RULES = { /** The program has dynamic-tier jsval operations (js.* builtins, the * jsval pass): links the scr_js_* units (two-tier design I3). */ dynTier: rule({ prefixes: ["js."] }), + /** Checked-dynamic stream slots test values and pipe into JavaScript + * stream objects (scr_stream_dyn.c). */ + streamDyn: rule({ fns: ["stream.dynNative", "stream.pipeDyn"] }), /** The stream unit emits through the emitter unit, also when the * program constructs no stream (a pipeline of functions). */ emitter: rule({ @@ -662,6 +669,7 @@ function noRuntimeFeatures(): RuntimeFeatures { searchParams: false, stdioExtended: false, stream: false, + streamDyn: false, symbol: false, tls: false, tlsCa: false, diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 63078ae4a..68997c295 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5224,6 +5224,21 @@ function validateFunction( } return true; } + if (e.fn === "stream.dynNative") { + const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined; + const side = e.args[1]; + const className = + side?.kind === "numLit" ? (side.value === 1 ? "%Readable" : "%Writable") : null; + const ok = + def && + className !== null && + e.args[0]?.type.kind === "dyn" && + def.arms.length === 2 && + def.arms.some((a) => a.kind === "object" && a.className === className) && + def.arms.some((a) => a.kind === "nullT"); + if (!ok) err(`libCall ${e.fn} must return its side's stream class or null`, e.loc); + return true; + } if (e.fn === "child.stdioStream") { const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined; const slot = e.args[1]; diff --git a/packages/compiler/src/native/sanitizer.test.ts b/packages/compiler/src/native/sanitizer.test.ts index 6ca84af6f..d9efb7cd8 100644 --- a/packages/compiler/src/native/sanitizer.test.ts +++ b/packages/compiler/src/native/sanitizer.test.ts @@ -76,6 +76,7 @@ test.skipIf(process.platform !== "linux" && process.platform !== "darwin")( youngNursery: false, eventTarget: false, dynTier: false, + streamDyn: false, warningReplace: false, stdioExtended: false, regex: true, diff --git a/packages/runtime-pack-common/runtime-pack-matrix.mjs b/packages/runtime-pack-common/runtime-pack-matrix.mjs index 1e959eccc..790864677 100644 --- a/packages/runtime-pack-common/runtime-pack-matrix.mjs +++ b/packages/runtime-pack-common/runtime-pack-matrix.mjs @@ -93,6 +93,8 @@ export const OPTIONAL_RUNTIME_SOURCES = [ ["scr_zlib_stream.c", all("zlib", "stream")], // The Readable operators await through the dyn-async unit. ["scr_stream_ops.c", all("stream", "dynAsync")], + // Checked-dynamic stream slots call JavaScript members through the dyn-invoke unit. + ["scr_stream_dyn.c", "streamDyn"], // fs.ReadStream / fs.WriteStream reference only the stream unit and base units. ["scr_fs_stream.c", "stream"], // FileHandle streams reference both the stream unit and FileHandle. diff --git a/packages/runtime/src/scr_stream_dyn.c b/packages/runtime/src/scr_stream_dyn.c new file mode 100644 index 000000000..f938f122d --- /dev/null +++ b/packages/runtime/src/scr_stream_dyn.c @@ -0,0 +1,349 @@ +/* Checked-dynamic stream slots (the compiler's dyn-streams.ts). + * + * A program whose NodeJS.ReadableStream / NodeJS.WritableStream slots also + * hold JavaScript stream objects compiles them as checked-dynamic values: a + * native stream rides its typed capsule, any other value is the JavaScript + * object itself. Member calls test the value here and take the native + * stream path or the value's own members. Linked only by such programs + * (the stream and dyn-invoke units together). */ +#include "scr_runtime.h" + +#include +#include + +static bool scr_stream_dyn_within(const ScrVt *vt, const ScrVt *cls) { + return vt == cls || (cls->pre <= vt->pre && vt->pre <= cls->post); +} + +/* The native stream a checked-dynamic value boxes (+1), when it has the + * requested side (1 readable, 2 writable; Duplex, Transform and PassThrough + * have both), or NULL for any other value. Never throws. */ +ScrStream *scr_stream_dyn_native(const ScrDyn *value, double side) { + if (value->kind != SCR_DYN_TYPED_REF || value->v.typed_ref.type_key_len <= 7 || + memcmp(value->v.typed_ref.type_key, "object:", 7) != 0) + return NULL; + ScrStream *stream = scr_dyn_typed_ref_unbox(value); + const ScrVt *vt = stream->vt; + bool duplex = scr_stream_dyn_within(vt, &scr_duplex_vt) || + scr_stream_dyn_within(vt, &scr_transform_vt) || + scr_stream_dyn_within(vt, &scr_passthrough_vt); + bool ok = duplex || (side == 1 ? scr_stream_dyn_within(vt, &scr_readable_vt) + : scr_stream_dyn_within(vt, &scr_writable_vt)); + if (ok) return stream; + value->v.typed_ref.release(stream); + return NULL; +} + +/* ── readable.pipe(destination) into a JavaScript stream object ───────── + * Node's Readable.prototype.pipe (lib/internal/streams/readable.js) over a + * native source and a destination that is not a native stream: the same + * listeners in the same order, with the destination's members called + * through the dyn-invoke unit. The context holds the source, its boxed + * value (what 'pipe' and 'unpipe' pass), the destination and the boxed + * listeners until cleanup; the live list breaks the listener cycles at + * exit without calling JavaScript. */ + +enum { + SCR_DYN_PIPE_DATA, + SCR_DYN_PIPE_END, + SCR_DYN_PIPE_UNPIPE, + SCR_DYN_PIPE_DRAIN, + SCR_DYN_PIPE_ERROR, + SCR_DYN_PIPE_CLOSE, + SCR_DYN_PIPE_FINISH, + SCR_DYN_PIPE_ON_UNPIPE, + SCR_DYN_PIPE_COUNT +}; + +typedef struct ScrDynPipe { + size_t rc; + ScrStream *src; /* +1 until dropped */ + ScrDyn *src_box; /* +1 until dropped */ + ScrDyn *dest; /* +1 until dropped */ + ScrDyn *l[SCR_DYN_PIPE_COUNT]; /* +1 boxed listeners, or NULL */ + bool end, cleaned, awaiting_drain, unpiped; + struct ScrDynPipe *live_prev, *live_next; +} ScrDynPipe; + +static SCR_TL ScrDynPipe *scr_dyn_pipe_live = NULL; + +/* A string value for an event name (+1). */ +static ScrDyn *scr_dyn_pipe_str(const char *text, size_t len) { + ScrStr *s = scr_str_new(text, len); + ScrDyn *value = scr_dyn_new_str(s); + scr_str_release(s); + return value; +} +static SCR_TL bool scr_dyn_pipe_atexit_set = false; + +static void *scr_dyn_pipe_retain_v(void *p) { + ((ScrDynPipe *)p)->rc++; + return p; +} + +static void scr_dyn_pipe_drop(ScrDynPipe *p) { + if (p->live_prev) p->live_prev->live_next = p->live_next; + else if (scr_dyn_pipe_live == p) scr_dyn_pipe_live = p->live_next; + if (p->live_next) p->live_next->live_prev = p->live_prev; + p->live_prev = p->live_next = NULL; + for (size_t i = 0; i < SCR_DYN_PIPE_COUNT; i++) { + ScrDyn *l = p->l[i]; + p->l[i] = NULL; + if (l) scr_dyn_release(l); + } + if (p->src) { + ScrStream *src = p->src; + p->src = NULL; + scr_stream_release(src); + } + if (p->src_box) { + ScrDyn *box = p->src_box; + p->src_box = NULL; + scr_dyn_release(box); + } + if (p->dest) { + ScrDyn *dest = p->dest; + p->dest = NULL; + scr_dyn_release(dest); + } +} + +static void scr_dyn_pipe_release_v(void *v) { + ScrDynPipe *p = v; + if (--p->rc > 0) return; + scr_dyn_pipe_drop(p); + free(p); +} + +static void scr_dyn_pipe_atexit(void) { + while (scr_dyn_pipe_live) { + ScrDynPipe *p = scr_dyn_pipe_retain_v(scr_dyn_pipe_live); + scr_dyn_pipe_drop(p); + scr_dyn_pipe_release_v(p); + } +} + +/* destination.method(args...), the result discarded; false on a throw. */ +static bool scr_dyn_pipe_call(ScrDynPipe *p, const char *method, ScrDyn *const *args, size_t argc) { + ScrDyn *result = scr_dyn_invoke(p->dest, method, args, argc, method); + if (result) scr_dyn_release(result); + return !scr_exc_pending(); +} + +static bool scr_dyn_pipe_dest_off(ScrDynPipe *p, const char *event, size_t which) { + if (!p->l[which]) return true; + ScrDyn *args[2] = {scr_dyn_pipe_str(event, strlen(event)), p->l[which]}; + bool ok = scr_dyn_pipe_call(p, "removeListener", args, 2); + scr_dyn_release(args[0]); + return ok; +} + +static void scr_dyn_pipe_src_off(ScrDynPipe *p, const char *event, size_t which) { + if (!p->l[which] || !p->src) return; + ScrStr *name = scr_str_new(event, strlen(event)); + scr_emitter_release(scr_emitter_off_dyn((ScrEmitter *)p->src, name, p->l[which])); + scr_str_release(name); +} + +static void scr_dyn_pipe_ondrain(ScrDynPipe *p) { + p->awaiting_drain = false; + if (!p->src) return; + ScrStr *data = scr_str_new("data", 4); + double listeners = scr_emitter_listener_count((ScrEmitter *)p->src, data); + scr_str_release(data); + if (listeners > 0) scr_stream_release(scr_stream_resume(p->src)); +} + +/* pipe's cleanup(): every listener off, then a pending drain resumes. */ +static void scr_dyn_pipe_cleanup(ScrDynPipe *p) { + if (p->cleaned) return; + if (!scr_dyn_pipe_dest_off(p, "close", SCR_DYN_PIPE_CLOSE) || + !scr_dyn_pipe_dest_off(p, "finish", SCR_DYN_PIPE_FINISH) || + !scr_dyn_pipe_dest_off(p, "drain", SCR_DYN_PIPE_DRAIN) || + !scr_dyn_pipe_dest_off(p, "error", SCR_DYN_PIPE_ERROR) || + !scr_dyn_pipe_dest_off(p, "unpipe", SCR_DYN_PIPE_ON_UNPIPE)) + return; + scr_dyn_pipe_src_off(p, "end", SCR_DYN_PIPE_END); + scr_dyn_pipe_src_off(p, "end", SCR_DYN_PIPE_UNPIPE); + scr_dyn_pipe_src_off(p, "data", SCR_DYN_PIPE_DATA); + p->cleaned = true; + if (p->l[SCR_DYN_PIPE_DRAIN] && p->awaiting_drain) scr_dyn_pipe_ondrain(p); + ScrDynPipe *held = scr_dyn_pipe_retain_v(p); + scr_dyn_pipe_drop(held); + scr_dyn_pipe_release_v(held); +} + +/* src.unpipe(dest): the source pauses (its only pipe goes) and the + * destination hears 'unpipe' with { hasUnpiped: false }. */ +static void scr_dyn_pipe_unpipe(ScrDynPipe *p) { + if (p->unpiped || !p->src) return; + p->unpiped = true; + scr_stream_release(scr_stream_pause(p->src)); + ScrDyn *info = scr_dyn_new_obj(); + scr_dyn_obj_set(info, "hasUnpiped", 10, scr_dyn_new_bool(false)); + ScrDyn *args[3] = {scr_dyn_pipe_str("unpipe", 6), scr_dyn_retain(p->src_box), info}; + (void)scr_dyn_pipe_call(p, "emit", args, 3); + for (size_t i = 0; i < 3; i++) scr_dyn_release(args[i]); +} + +static ScrDyn *scr_dyn_pipe_listener(ScrDynPipe *p, int kind); + +/* pipe's pause(): the source stops until the destination drains. */ +static void scr_dyn_pipe_pause(ScrDynPipe *p) { + if (!p->cleaned) { + p->awaiting_drain = true; + scr_stream_release(scr_stream_pause(p->src)); + } + if (!p->l[SCR_DYN_PIPE_DRAIN] && p->dest) { + p->l[SCR_DYN_PIPE_DRAIN] = scr_dyn_pipe_listener(p, SCR_DYN_PIPE_DRAIN); + ScrDyn *args[2] = {scr_dyn_pipe_str("drain", 5), p->l[SCR_DYN_PIPE_DRAIN]}; + (void)scr_dyn_pipe_call(p, "on", args, 2); + scr_dyn_release(args[0]); + } +} + +static ScrDyn *scr_dyn_pipe_thunk(ScrClosure *clo, ScrDyn *const *args, size_t argc) { + ScrDynPipe *p = scr_box_get_ref(clo->caps[0]); /* +1 */ + int kind = (int)scr_box_get_f64(clo->caps[1]); + ScrDyn *first = argc ? args[0] : scr_dyn_undefined(); + switch (kind) { + case SCR_DYN_PIPE_DATA: { + if (!p->dest) break; + ScrDyn *ret = scr_dyn_invoke(p->dest, "write", &first, 1, "write"); + if (ret) { + if (ret->kind == SCR_DYN_BOOL && !scr_dyn_truthy(ret)) scr_dyn_pipe_pause(p); + scr_dyn_release(ret); + } + break; + } + case SCR_DYN_PIPE_END: + if (p->dest) (void)scr_dyn_pipe_call(p, "end", NULL, 0); + break; + case SCR_DYN_PIPE_UNPIPE: + scr_dyn_pipe_unpipe(p); + break; + case SCR_DYN_PIPE_DRAIN: + scr_dyn_pipe_ondrain(p); + break; + case SCR_DYN_PIPE_ERROR: { + if (!p->dest) break; + ScrDyn *dest = scr_dyn_retain(p->dest); + ScrDyn *self = scr_dyn_retain(p->l[SCR_DYN_PIPE_ERROR]); + scr_dyn_pipe_unpipe(p); + if (!scr_exc_pending()) { + ScrDyn *off[2] = {scr_dyn_pipe_str("error", 5), self}; + ScrDyn *r = scr_dyn_invoke(dest, "removeListener", off, 2, "removeListener"); + if (r) scr_dyn_release(r); + ScrDyn *count = scr_exc_pending() ? NULL : scr_dyn_invoke(dest, "listenerCount", off, 1, "listenerCount"); + if (count && count->kind == SCR_DYN_NUM && !scr_dyn_truthy(count)) { + ScrDyn *emit[2] = {off[0], first}; + r = scr_dyn_invoke(dest, "emit", emit, 2, "emit"); + if (r) scr_dyn_release(r); + } + if (count) scr_dyn_release(count); + scr_dyn_release(off[0]); + } + scr_dyn_release(self); + scr_dyn_release(dest); + break; + } + case SCR_DYN_PIPE_CLOSE: + if (scr_dyn_pipe_dest_off(p, "finish", SCR_DYN_PIPE_FINISH)) scr_dyn_pipe_unpipe(p); + break; + case SCR_DYN_PIPE_FINISH: + if (scr_dyn_pipe_dest_off(p, "close", SCR_DYN_PIPE_CLOSE)) scr_dyn_pipe_unpipe(p); + break; + case SCR_DYN_PIPE_ON_UNPIPE: { + ScrDyn *info = argc > 1 ? args[1] : NULL; + if (p->src_box && scr_dyn_strict_eq(first, p->src_box) && info && info->kind == SCR_DYN_OBJ) { + ScrDyn *has = scr_dyn_obj_read(info, "hasUnpiped", 10); + bool unpiped = has->kind == SCR_DYN_BOOL && !scr_dyn_truthy(has); + scr_dyn_release(has); + if (unpiped) { + scr_dyn_obj_set(info, "hasUnpiped", 10, scr_dyn_new_bool(true)); + scr_dyn_pipe_cleanup(p); + } + } + break; + } + } + scr_dyn_pipe_release_v(p); + return scr_exc_pending() ? NULL : scr_dyn_retain(scr_dyn_undefined()); +} + +static ScrDyn *scr_dyn_pipe_listener(ScrDynPipe *p, int kind) { + ScrClosure *clo = scr_closure_new((void *)scr_dyn_pipe_thunk, 2); + clo->caps[0] = scr_box_new_obj(scr_dyn_pipe_retain_v, scr_dyn_pipe_release_v, NULL); + scr_box_set_ref(clo->caps[0], scr_dyn_pipe_retain_v(p)); + clo->caps[1] = scr_box_new(SCR_BOX_F64); + scr_box_set_f64(clo->caps[1], (double)kind); + return scr_dyn_new_func(clo, scr_dyn_pipe_thunk, 1, "native pipe listener", NULL); +} + +/* destination.on / once / prependListener(event, listener) */ +static bool scr_dyn_pipe_listen(ScrDynPipe *p, const char *method, const char *event, int kind) { + p->l[kind] = scr_dyn_pipe_listener(p, kind); + ScrDyn *args[2] = {scr_dyn_pipe_str(event, strlen(event)), p->l[kind]}; + bool ok = scr_dyn_pipe_call(p, method, args, 2); + scr_dyn_release(args[0]); + return ok; +} + +static void scr_dyn_pipe_src_on(ScrDynPipe *p, const char *event, int kind, bool once) { + p->l[kind] = scr_dyn_pipe_listener(p, kind); + ScrStr *name = scr_str_new(event, strlen(event)); + scr_stream_release(scr_stream_on_dyn(p->src, name, p->l[kind], once, false)); + scr_str_release(name); +} + +/* readable.pipe(destination, { end }) for a destination that is not a + * native stream. src and src_box (the source's boxed value) and dest are + * borrowed. Throws what the destination's members throw. */ +void scr_stream_pipe_dyn(ScrStream *src, ScrDyn *src_box, ScrDyn *dest, bool end) { + ScrDynPipe *p = calloc(1, sizeof *p); + if (!p) abort(); + p->rc = 1; + p->src = scr_stream_retain(src); + p->src_box = scr_dyn_retain(src_box); + p->dest = scr_dyn_retain(dest); + p->end = end; + p->live_next = scr_dyn_pipe_live; + if (scr_dyn_pipe_live) scr_dyn_pipe_live->live_prev = p; + scr_dyn_pipe_live = p; + if (!scr_dyn_pipe_atexit_set) { + scr_dyn_pipe_atexit_set = true; + scr_atexit(scr_dyn_pipe_atexit); + } + /* The source's end ends the destination (end: true) or unpipes it. */ + scr_dyn_pipe_src_on(p, "end", end ? SCR_DYN_PIPE_END : SCR_DYN_PIPE_UNPIPE, true); + if (scr_exc_pending() || !scr_dyn_pipe_listen(p, "on", "unpipe", SCR_DYN_PIPE_ON_UNPIPE)) goto out; + scr_dyn_pipe_src_on(p, "data", SCR_DYN_PIPE_DATA, false); + if (scr_exc_pending()) goto out; + { + ScrDyn *prepend = scr_dyn_obj_read(dest, "prependListener", 15); + bool has_prepend = prepend->kind == SCR_DYN_FUNC; + scr_dyn_release(prepend); + if (!scr_dyn_pipe_listen(p, has_prepend ? "prependListener" : "on", "error", SCR_DYN_PIPE_ERROR)) + goto out; + } + if (!scr_dyn_pipe_listen(p, "once", "close", SCR_DYN_PIPE_CLOSE) || + !scr_dyn_pipe_listen(p, "once", "finish", SCR_DYN_PIPE_FINISH)) + goto out; + { + ScrDyn *args[2] = {scr_dyn_pipe_str("pipe", 4), scr_dyn_retain(src_box)}; + bool ok = scr_dyn_pipe_call(p, "emit", args, 2); + scr_dyn_release(args[0]); + scr_dyn_release(args[1]); + if (!ok) goto out; + } + { + ScrDyn *need = scr_dyn_obj_read(dest, "writableNeedDrain", 17); + bool pause = need->kind == SCR_DYN_BOOL && scr_dyn_truthy(need); + scr_dyn_release(need); + if (pause) scr_dyn_pipe_pause(p); + else if (scr_stream_is_paused(src)) + scr_stream_release(scr_stream_resume(src)); + } +out: + scr_dyn_pipe_release_v(p); +}