From 19091ebbda4276f46b34d0f2eef526f639c0e843 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Tue, 20 Oct 2026 14:49:30 -0500 Subject: [PATCH 1/2] Pipe into constructor-function streams natively in checked-dynamic stream slots With checked-dynamic stream slots, stream.dynNative answered only boxed runtime streams, so a pipe into a constructor-function stream took the JavaScript listener protocol, and src.unpipe(this) in its 'pipe' listener had no runtime pipe to remove (6564-constructor-function-stream-late-hooks). In programs that link the constructor-function stream unit, stream.dynNative now answers such a stream's runtime stream, as stream.unboxDyn does there. A stream that overrides write() or end() still pipes through its JavaScript members. --- .../compiler/src/backend/llvm/lib-stream.ts | 11 +++++--- packages/runtime/src/scr_dyn.h | 4 +++ packages/runtime/src/scr_dyn_builtin_stream.c | 25 +++++++++++++++++++ 3 files changed, 36 insertions(+), 4 deletions(-) diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index 707fb6dbca..ae9b6269c2 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -47,11 +47,14 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV 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)`); + // Programs with constructor-function streams answer those streams too + // (scr_dyn_builtin_stream.c), as stream.unboxDyn does there. + const fn = host.usesDynBuiltinStream + ? "scr_dyn_builtin_stream_dyn_native" + : "scr_stream_dyn_native"; + host.declare(`declare ptr @${fn}(ptr, double)`); const raw = B.tmp(); - B.line( - `${raw} = call ptr @scr_stream_dyn_native(ptr ${value.name}, double ${side.name}) ; +1 or NULL`, - ); + B.line(`${raw} = call ptr @${fn}(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") { diff --git a/packages/runtime/src/scr_dyn.h b/packages/runtime/src/scr_dyn.h index 20392e3a51..401bf2b392 100644 --- a/packages/runtime/src/scr_dyn.h +++ b/packages/runtime/src/scr_dyn.h @@ -1192,6 +1192,10 @@ ScrDyn *scr_dyn_builtin_stream_object(struct ScrStream *s); ScrDyn *scr_dyn_builtin_stream_view(struct ScrStream *s); struct ScrStream *scr_dyn_builtin_stream_check(const ScrDyn *value, double kind); struct ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *message); +/* stream.dynNative in those programs: scr_stream_dyn_native's contract, + * also taking constructor-function streams that keep the runtime's write + * and end (+1 or NULL). */ +struct ScrStream *scr_dyn_builtin_stream_dyn_native(const ScrDyn *value, double side); bool scr_dyn_class_is_key(const ScrDyn *value, const ScrStr *type_key); bool scr_dyn_class_is(const ScrDyn *value, const char *type_key); /* Calls a dyn value: a non-function kind throws the catchable TypeError diff --git a/packages/runtime/src/scr_dyn_builtin_stream.c b/packages/runtime/src/scr_dyn_builtin_stream.c index 1926b0622c..b7295f59aa 100644 --- a/packages/runtime/src/scr_dyn_builtin_stream.c +++ b/packages/runtime/src/scr_dyn_builtin_stream.c @@ -1673,6 +1673,31 @@ ScrStream *scr_dyn_builtin_stream_check(const ScrDyn *value, double kind) { return ok ? scr_stream_retain(s) : NULL; } +/* stream.dynNative (checked-dynamic stream slots, scr_stream_dyn.c) in + * programs that link this unit: a constructor-function stream with the + * requested side (1 readable, 2 writable) answers its runtime stream (+1), + * so a pipe into it, and an unpipe of it, are the runtime's own, unless it + * overrides write() or end(), which the runtime's pipe would bypass: that + * one is NULL and pipes through its JavaScript members. Other values take + * scr_stream_dyn_native. Never throws. */ +ScrStream *scr_dyn_builtin_stream_dyn_native(const ScrDyn *value, double side) { + ScrStream *s = value->kind == SCR_DYN_OBJ ? bcs_of(value) : NULL; + if (s) { + const ScrVt *vt = s->vt; + bool readable = vt->pre >= scr_readable_vt.pre && vt->pre <= scr_readable_vt.post; + bool writable = vt->pre >= scr_writable_vt.pre && vt->pre <= scr_writable_vt.post; + bool duplex = vt->pre >= scr_duplex_vt.pre && vt->pre <= scr_duplex_vt.post; + if (!(duplex || (side == 1 ? readable : writable))) return NULL; + ScrDyn *object = (ScrDyn *)value; + if (side == 2 && (bcs_overrides(object, "write") || bcs_overrides(object, "end"))) { + if (scr_exc_pending()) scr_exc_clear(); + return NULL; + } + return scr_stream_retain(s); + } + return scr_stream_dyn_native(value, side); +} + /* stream.unboxDyn (an untyped pipe destination or pipeline stage) in * programs that link this unit: a constructor-function stream answers its * runtime stream (+1), unless it overrides write() or end(), which the From 3d280218b3c118bec1d00f8a015ebcb338b7b1ef Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Tue, 20 Oct 2026 15:58:30 -0500 Subject: [PATCH 2/2] Keep constructor-function stream methods on stream slot member calls; order 6471's file reads Only pipe and unpipe destinations now take the constructor-function stream probe (stream.dynNativePipe). Member calls on a checked-dynamic stream slot use the boxed-native probe again, so a constructor-function stream's own pause(), resume() and on() still run (5820-node-stream-dyn-constructor-overrides). 6471 started its second file read on the first stream's 'end', so that stream's 'close' could land after the second stream's data. The second read now starts on 'close'. --- .../compiler/src/backend/llvm/lib-stream.ts | 14 ++++--- .../src/frontend/lowering/dyn-streams.ts | 18 ++++++--- .../compiler/src/ir/builtin-signatures.ts | 4 ++ packages/compiler/src/ir/runtime-features.ts | 6 ++- packages/compiler/src/ir/validate.ts | 2 +- .../main.ts.json | 7 ++++ .../main.ts | 35 +++++++++++++++++ .../sources.js | 39 +++++++++++++++++++ .../6471-stream-data-buffer-or-string.ts | 7 +++- .../main.ts.json | 1 + 10 files changed, 118 insertions(+), 15 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts.json create mode 100644 tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts create mode 100644 tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js create mode 100644 tests/harness/corpus-costs/5820-node-stream-dyn-constructor-overrides/main.ts.json diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index ae9b6269c2..a319d02574 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -36,7 +36,7 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV host.emitPendingCheck(); return out; } - if (e.fn === "stream.dynNative") { + if (e.fn === "stream.dynNative" || e.fn === "stream.dynNativePipe") { // 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`); @@ -47,11 +47,13 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV 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]!); - // Programs with constructor-function streams answer those streams too - // (scr_dyn_builtin_stream.c), as stream.unboxDyn does there. - const fn = host.usesDynBuiltinStream - ? "scr_dyn_builtin_stream_dyn_native" - : "scr_stream_dyn_native"; + // A pipe or unpipe destination in programs with constructor-function + // streams can be one of those streams too (scr_dyn_builtin_stream.c), + // as stream.unboxDyn does there. + const fn = + e.fn === "stream.dynNativePipe" && host.usesDynBuiltinStream + ? "scr_dyn_builtin_stream_dyn_native" + : "scr_stream_dyn_native"; host.declare(`declare ptr @${fn}(ptr, double)`); const raw = B.tmp(); B.line(`${raw} = call ptr @${fn}(ptr ${value.name}, double ${side.name}) ; +1 or NULL`); diff --git a/packages/compiler/src/frontend/lowering/dyn-streams.ts b/packages/compiler/src/frontend/lowering/dyn-streams.ts index 51bcf0ea89..2ac09e8e59 100644 --- a/packages/compiler/src/frontend/lowering/dyn-streams.ts +++ b/packages/compiler/src/frontend/lowering/dyn-streams.ts @@ -58,11 +58,16 @@ export function nativeStreamClass(side: "r" | "w"): IrType { /** 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 { +function dynNativeStream( + lowerer: Lowerer, + value: IrExpr, + side: "r" | "w", + fn: "stream.dynNative" | "stream.dynNativePipe" = "stream.dynNative", +): IrExpr { const native = nativeStreamClass(side); return { kind: "libCall", - fn: "stream.dynNative", + fn, args: [value, numLit(side === "r" ? 1 : 2, value.loc)], type: { kind: "union", unionId: lowerer.unions.intern([native, { kind: "nullT" }]) }, loc: value.loc, @@ -268,8 +273,11 @@ export function lowerForAwaitDynStream( } /** 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. */ + * checked-dynamic pipe or unpipe destination stands for (typed as the + * side's class), or `otherwise` for any other value. The destination's + * stream is a boxed native one or a constructor-function stream's + * (stream.dynNativePipe): only the runtime's pipe reads it, while member + * calls on a slot keep a constructor-function stream's own methods. */ export function dynNativeBranch( lowerer: Lowerer, value: IrExpr, @@ -279,7 +287,7 @@ export function dynNativeBranch( ): IrStmt { const loc = value.loc; const nativeType = nativeStreamClass(side); - const nullable = dynNativeStream(lowerer, value, side); + const nullable = dynNativeStream(lowerer, value, side, "stream.dynNativePipe"); const nativeOrNull = lowerer.declareHiddenLocal("%streamNative", nullable.type); const unionId = nullable.type.kind === "union" ? nullable.type.unionId : ""; const tag = lowerer.armTag(unionId, nativeType); diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 1ffae75360..cd3db784dc 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -2689,6 +2689,10 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ * (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 }, + /** stream.dynNative for a pipe or unpipe destination: also a + * constructor-function stream's runtime stream, in programs that link + * that unit, unless it overrides write() or end(). */ + "stream.dynNativePipe": { 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 diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 808b20fe44..fd7c98cf6e 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -170,6 +170,7 @@ const STREAM_LIB_FNS: ReadonlySet = new Set([ "stream.errored", "stream.prop", "stream.dynNative", + "stream.dynNativePipe", "stream.pipeDyn", "stream.unpipeDyn", "stream.stdioWritable", @@ -209,6 +210,7 @@ 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.dynNativePipe", "stream.pipeDyn", "stream.unpipeDyn", "dyn.arrayProtoCall", @@ -568,7 +570,9 @@ const RUNTIME_FEATURE_RULES = { 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", "stream.unpipeDyn"] }), + streamDyn: rule({ + fns: ["stream.dynNative", "stream.dynNativePipe", "stream.pipeDyn", "stream.unpipeDyn"], + }), /** The stream unit emits through the emitter unit, also when the * program constructs no stream (a pipeline of functions). */ emitter: rule({ diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 2a2ddbd2f1..f75f6f46e3 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5233,7 +5233,7 @@ function validateFunction( } return true; } - if (e.fn === "stream.dynNative") { + if (e.fn === "stream.dynNative" || e.fn === "stream.dynNativePipe") { const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined; const side = e.args[1]; const className = diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts.json new file mode 100644 index 0000000000..49f4052b19 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts.json @@ -0,0 +1,7 @@ +{ + "order": [ + "/tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js", + "/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts" + ], + "diags": [] +} diff --git a/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts b/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts new file mode 100644 index 0000000000..223e10ee86 --- /dev/null +++ b/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts @@ -0,0 +1,35 @@ +// Constructor-function streams in NodeJS.ReadableStream / WritableStream +// slots: member calls run the stream's own pause, resume and on, and a pipe +// into one, and an unpipe of it, are the runtime's. +import { Readable } from "node:stream"; +import { collector, recordingSource } from "./sources.js"; + +function control(stream: NodeJS.ReadableStream): void { + stream.pause(); + stream.resume(); + stream.on("end", () => {}); +} + +function attach(src: Readable, dst: NodeJS.WritableStream): void { + src.pipe(dst); +} + +function detach(src: Readable, dst: NodeJS.WritableStream): void { + src.unpipe(dst); +} + +const recording = recordingSource(); +control(recording); +console.log("recorded", recording.calls.join(",")); + +const sink = collector(); +sink.on("pipe", () => console.log("pipe event")); +sink.on("unpipe", () => console.log("unpipe event")); +const source = new Readable({ read() {} }); +attach(source, sink); +source.push("a"); +setImmediate(() => { + detach(source, sink); + source.push("b"); + setImmediate(() => console.log("collected", sink.chunks.join(","), source.isPaused())); +}); diff --git a/tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js b/tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js new file mode 100644 index 0000000000..462cd05387 --- /dev/null +++ b/tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js @@ -0,0 +1,39 @@ +// Constructor-function streams that override stream methods, and a plain +// JavaScript sink. +import { Readable, Writable } from "node:stream"; +import util from "node:util"; + +/** A Readable whose pause, resume and on record their calls before + * running Readable's own. */ +function Recording() { + Readable.call(this, { read() {} }); + this.calls = []; +} +util.inherits(Recording, Readable); +for (const name of ["pause", "resume", "on"]) { + Recording.prototype[name] = function (...args) { + this.calls.push(name); + return Readable.prototype[name].apply(this, args); + }; +} + +/** @returns {any} */ +export function recordingSource() { + return new Recording(); +} + +/** A Writable without overrides, piped into natively. */ +function Collector() { + Writable.call(this); + this.chunks = []; +} +util.inherits(Collector, Writable); +Collector.prototype._write = function (chunk, _encoding, done) { + this.chunks.push(String(chunk)); + done(); +}; + +/** @returns {any} */ +export function collector() { + return new Collector(); +} diff --git a/tests/corpus/6471-stream-data-buffer-or-string.ts b/tests/corpus/6471-stream-data-buffer-or-string.ts index 69289ced57..54c63945b6 100644 --- a/tests/corpus/6471-stream-data-buffer-or-string.ts +++ b/tests/corpus/6471-stream-data-buffer-or-string.ts @@ -32,9 +32,12 @@ fs.writeFileSync(file, "file contents\n"); const stream = fs.createReadStream(file); const release = (): void => console.log("released"); stream.on("data", (chunk: Buffer | string) => describe("file", chunk)); -stream.on("close", release); stream.on("error", release); -stream.on("end", () => { +// The second read starts once the first stream has closed its file, so the +// order of its 'close' and the second stream's data does not depend on +// which file operation finishes first. +stream.on("close", () => { + release(); const encoded = fs.createReadStream(file, { encoding: "utf8" }); encoded.on("data", (chunk: Buffer | string) => describe("file utf8", chunk)); encoded.on("close", () => { diff --git a/tests/harness/corpus-costs/5820-node-stream-dyn-constructor-overrides/main.ts.json b/tests/harness/corpus-costs/5820-node-stream-dyn-constructor-overrides/main.ts.json new file mode 100644 index 0000000000..f8ccf06992 --- /dev/null +++ b/tests/harness/corpus-costs/5820-node-stream-dyn-constructor-overrides/main.ts.json @@ -0,0 +1 @@ +{ "plain": 3490, "san": 8050 }