Fix Readable.push and write of string | null and Buffer | null chunks (#1698)

packages/compiler/src/backend/llvm/lib-stream.ts

@@ -274,10 +274,9 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
       .filter((a) => a.tag >= 0);
     if (present.length === 0)
       throw new InternalCompilerError(`llvm emitter bug: ${e.fn} union lacks its arms`);
-    const tagP = B.tmp();
-    const tag = B.tmp();
-    B.line(`${tagP} = getelementptr inbounds %ScrUnion, ptr ${args[1]!.name}, i64 0, i32 1`);
-    B.line(`${tag} = load i32, ptr ${tagP}`);
+    // A tagged box, or the arm pointer itself for a nullable union
+    // (`string | null`, `Buffer | null`).
+    const tag = host.unionTag(args[1]!.name, t.unionId);
     const slot = B.slot();
     B.entryAllocas.push(`${slot} = alloca i1`);
     const lj = B.newLabel("scu.j");
@@ -290,10 +289,7 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
         B.line(`store i1 ${r}, ptr ${slot}`);
         return;
       }
-      const pk = B.tmp();
-      const pv = B.tmp();
-      B.line(`${pk} = getelementptr inbounds %ScrUnion, ptr ${args[1]!.name}, i64 0, i32 5`);
-      B.line(`${pv} = load ptr, ptr ${pk} ; borrowed payload`);
+      const pv = host.unionPeek(args[1]!.name, t.unionId); // borrowed payload
       const r = B.tmp();
       if (pushing) {
         host.declare(`declare zeroext i1 @${entry}(ptr, ptr)`);

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6820-stream-push-nullable-chunks.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6820-stream-push-nullable-chunks.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6821-stream-write-nullable-chunks.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6821-stream-write-nullable-chunks.cjs"
+ ],
+ "diags": []
+}

tests/corpus/6820-stream-push-nullable-chunks.ts

@@ -0,0 +1,77 @@
+// Readable subclasses pushing `string | null` and `Buffer | null` chunks, a
+// conditional `push(done ? null : chunk)`, and writes of `string | Buffer`
+// chunks.
+import { Readable, Writable } from "node:stream";
+
+class Lines extends Readable {
+  private index = 0;
+  private readonly lines: string[];
+  constructor(lines: string[]) {
+    super();
+    this.lines = lines;
+  }
+  next(): string | null {
+    return this.index < this.lines.length ? this.lines[this.index++]! : null;
+  }
+  _read(): void {
+    const line: string | null = this.next();
+    this.push(line);
+  }
+}
+
+class Blocks extends Readable {
+  private sent = 0;
+  _read(): void {
+    const block: Buffer | null = this.sent < 3 ? Buffer.from([65 + this.sent, 10]) : null;
+    this.sent++;
+    this.push(block);
+  }
+}
+
+class Countdown extends Readable {
+  private left: number;
+  constructor(left: number) {
+    super({ encoding: "utf8" });
+    this.left = left;
+  }
+  _read(): void {
+    this.left--;
+    this.push(this.left < 0 ? null : `t-${this.left} `);
+  }
+}
+
+function collect(label: string, stream: Readable): Promise<void> {
+  return new Promise((resolve) => {
+    const parts: string[] = [];
+    stream.on("data", (chunk: Buffer | string) =>
+      parts.push(typeof chunk === "string" ? chunk : chunk.toString()),
+    );
+    stream.on("end", () => {
+      console.log(label, JSON.stringify(parts.join("")), parts.length);
+      resolve();
+    });
+  });
+}
+
+async function main(): Promise<void> {
+  await collect("lines", new Lines(["a\n", "bb\n", "ccc\n"]));
+  await collect("empty lines", new Lines([]));
+  await collect("blocks", new Blocks());
+  await collect("countdown", new Countdown(3));
+
+  const written: string[] = [];
+  const sink = new Writable({
+    write(chunk: Buffer, _encoding, callback) {
+      written.push(chunk.toString());
+      callback();
+    },
+  });
+  const pieces: (string | Buffer)[] = ["x", Buffer.from("y"), "z"];
+  for (const piece of pieces) {
+    const chunk: string | Buffer = piece;
+    console.log("write", Buffer.isBuffer(chunk) ? "buffer" : "string", sink.write(chunk));
+  }
+  sink.end(() => console.log("written", JSON.stringify(written.join(""))));
+}
+
+void main();

tests/corpus/6821-stream-write-nullable-chunks.cjs

@@ -0,0 +1,23 @@
+// Writes and pushes of a chunk that is a string or null: a null write throws
+// ERR_STREAM_NULL_VALUES, and a null push ends the readable.
+const { Writable, Readable } = require("node:stream");
+const written = [];
+const sink = new Writable({
+  write(chunk, _encoding, callback) {
+    written.push(chunk.toString());
+    callback();
+  },
+});
+for (let i = 0; i < 3; i++) {
+  const chunk = i === 1 ? null : `c${i}`;
+  try {
+    console.log("write", JSON.stringify(chunk), sink.write(chunk));
+  } catch (error) {
+    console.log("write threw", error.name, error.code, error.message);
+  }
+}
+const r = new Readable({ read() {} });
+for (let i = 0; i < 3; i++) r.push(i === 2 ? null : `r${i}`);
+r.on("data", (d) => console.log("data", String(d)));
+r.on("end", () => console.log("end"));
+sink.end(() => console.log("written", JSON.stringify(written.join(""))));

tests/harness/corpus-costs/6820-stream-push-nullable-chunks.ts.json

@@ -0,0 +1 @@
+{ "plain": 3100, "san": 41100 }

tests/harness/corpus-costs/6821-stream-write-nullable-chunks.cjs.json

@@ -0,0 +1 @@
+{ "plain": 2400, "san": 40400 }