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'.

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

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": []
+}

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 }