Pipe into constructor-function streams natively in checked-dynamic stream slots

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,16 @@ 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)`);
+    // 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 @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") {

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);

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

packages/compiler/src/ir/runtime-features.ts

@@ -170,6 +170,7 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = 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<IrLibFn> = 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({

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 =

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts.json

@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/5820-node-stream-dyn-constructor-overrides/sources.js",
+  "<repo>/tests/corpus/5820-node-stream-dyn-constructor-overrides/main.ts"
+ ],
+ "diags": []
+}

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

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

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()));
+});

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();
+}

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", () => {

tests/harness/corpus-costs/5820-node-stream-dyn-constructor-overrides/main.ts.json

@@ -0,0 +1 @@
+{ "plain": 3490, "san": 8050 }