Keep one process-stream Writable per descriptor and never end it from a pipe

Under the general stream representation, every conversion of
process.stdout or process.stderr into a Writable slot made a new stream,
so two NodeJS.WritableStream slots holding process.stdout compared
unequal and kept separate listeners and writable state. A pipe into such
a slot also ended the converted Writable at the source's end, so later
writes through the slot failed; Node never ends process.stdout or
process.stderr from a pipe.

scr_stream_stdio_writable now keeps one Writable per descriptor and
thread, released at exit. Under the general representation, a pipe into
a Writable slot lowers to readable.pipeGeneral (scr_stream_pipe_general,
last in the stream unit), which keeps end off when the destination is
one of those Writables; programs without the general representation
lower pipes as before.

The fallback stream.Writable's on() and once() overloads for drain,
finish, close and error return the stream again, so listener chains such
as out.on("finish", listener).end(data) type-check.

Adds corpus 6468.

packages/compiler/ambient/scriptc-node-fallback.d.ts

@@ -6736,12 +6736,12 @@ declare module "stream" {
     readonly closed: boolean;
     readonly destroyed: boolean;
     readonly errored: Error | null;
-    on(event: "drain" | "finish" | "close", listener: () => void): void;
-    on(event: "error", listener: (err: Error) => void): void;
+    on(event: "drain" | "finish" | "close", listener: () => void): this;
+    on(event: "error", listener: (err: Error) => void): this;
     on(event: "pipe" | "unpipe", listener: (src: Readable) => void): this;
     on(event: string, listener: (...args: any[]) => void): this;
-    once(event: "drain" | "finish" | "close", listener: () => void): void;
-    once(event: "error", listener: (err: Error) => void): void;
+    once(event: "drain" | "finish" | "close", listener: () => void): this;
+    once(event: "error", listener: (err: Error) => void): this;
     once(event: "pipe" | "unpipe", listener: (src: Readable) => void): this;
     once(event: string, listener: (...args: any[]) => void): this;
     addListener(event: "drain" | "finish" | "close", listener: () => void): this;

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

@@ -1106,6 +1106,7 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "readable.opFind": "scr_stream_op_find",
   "readable.opReduce": "scr_stream_op_reduce",
   "readable.pipe": "scr_stream_pipe",
+  "readable.pipeGeneral": "scr_stream_pipe_general",
   "writable.cork": "scr_stream_cork",
   "writable.uncork": "scr_stream_uncork",
   "stream.destroyErr": "scr_stream_destroy",
@@ -1588,6 +1589,7 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "readable.opReduce",
   "readable.resume",
   "readable.pipe",
+  "readable.pipeGeneral",
   "readable.unpipe",
   "writable.write",
   "writable.writeStr",

packages/compiler/src/frontend/lowering/lower-stream.ts

@@ -3371,6 +3371,16 @@ function pipeIntoStdio(
   };
 }
 
+/** The pipe() libCall for a destination of type `dst`. Under the general
+ * stream representation a Writable slot can hold the converted
+ * process.stdout / process.stderr, which a pipe never ends (as in Node), so
+ * the pipe takes readable.pipeGeneral, which checks for them. */
+function generalPipe(lowerer: Lowerer, dst: IrType): "readable.pipe" | "readable.pipeGeneral" {
+  return lowerer.generalNodeStreams && dst.kind === "object" && dst.className === "%Writable"
+    ? "readable.pipeGeneral"
+    : "readable.pipe";
+}
+
 /** `readable.pipe(untyped)`: a checked-dynamic destination unwraps a boxed
  * native stream at run time (stream.unboxDyn; other values throw the
  * not-supported Error), and pipe() answers the destination value itself. */
@@ -3394,7 +3404,7 @@ function pipeIntoDyn(
     kind: "exprStmt",
     expr: {
       kind: "libCall",
-      fn: "readable.pipe",
+      fn: lowerer.generalNodeStreams ? "readable.pipeGeneral" : "readable.pipe",
       args: [
         { kind: "varRef", localId: source.id, type: receiverValue.type, loc },
         {
@@ -3691,7 +3701,7 @@ export function lowerStreamMethodCall(
     const receiver = lowerer.lowerExpr(access.expression);
     return {
       kind: "libCall",
-      fn: "readable.pipe",
+      fn: generalPipe(lowerer, dst.type),
       args: [receiver, dst, end],
       type: dst.type,
       loc,

packages/compiler/src/ir/builtin-effects.ts

@@ -326,6 +326,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.onDyn",
   "readable.resume",
   "readable.pipe",
+  "readable.pipeGeneral",
   "readable.unpipe",
   "writable.write",
   "writable.writeStr",

packages/compiler/src/ir/builtin-signatures.ts

@@ -3612,7 +3612,9 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * returns the below-hwm answer. readable.read: (recv, size — -1 for
    * absent) → Buffer|null union. pause/resume return recv +1 (`this`);
    * isPaused answers the flag. readable.pipe: (recv, dst, end) → dst +1,
-   * fires 'pipe' on dst; readable.unpipe (recv[, dst]) → recv +1.
+   * fires 'pipe' on dst; readable.pipeGeneral: the same, with `end` off
+   * when dst is the converted process.stdout / process.stderr;
+   * readable.unpipe (recv[, dst]) → recv +1.
    * writable.write/writeStr: (recv, chunk[, cb]) → below-hwm bool (cb
    * MOVES when present, called after the user write completes).
    * writable.end: (recv, flags[, chunk][, cb]) → recv +1. cork/uncork are
@@ -3810,6 +3812,7 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
   "readable.resume": { argTypes: [null], result: VOID },
   "readable.isPaused": { argTypes: [null], result: BOOL },
   "readable.pipe": { argTypes: [null, null, BOOL], result: VOID },
+  "readable.pipeGeneral": { argTypes: [null, null, BOOL], result: VOID },
   "readable.unpipe": { argTypes: [null], result: VOID },
   "readable.flowing": { argTypes: [null], result: VOID },
   "writable.write": { argTypes: [null, BYTES_U8], result: BOOL },

packages/compiler/src/ir/validate.ts

@@ -5729,9 +5729,9 @@ function validateFunction(
         }
         return true;
       }
-      if (e.fn === "readable.pipe") {
+      if (e.fn === "readable.pipe" || e.fn === "readable.pipeGeneral") {
         if (!isStreamObject(e.args[1]?.type) || !typeEquals(e.type, e.args[1]!.type)) {
-          err(`libCall readable.pipe must return its destination's type`, e.loc);
+          err(`libCall ${e.fn} must return its destination's type`, e.loc);
         }
         return true;
       }

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6468-node-stdio-writable-slot.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_stream.c

@@ -6665,18 +6665,42 @@ ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len)
  * stream.stdioWritable libCall): a Writable whose _write hands each chunk
  * to the descriptor through process.stdout.write's synchronous path and
  * completes at once, so ordering against console output and direct writes
- * is unchanged. The closure keeps the fd as given (1 or 2). */
+ * is unchanged. The closure keeps the fd as given (1 or 2). Each thread
+ * keeps one per descriptor, so every conversion of process.stdout answers
+ * the same stream, as Node's process.stdout is one object: listeners,
+ * writable state and identity are shared. The cache is released at exit. */
+static SCR_TL ScrStream *scr_stream_stdio_writables[2];
 static void scr_stream_stdio_write(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) {
   int fd = (int)scr_box_get_f64(cb->caps[0]);
   (void)scr_stdio_write(fd == 2 ? 2 : 1, chunk->data, chunk->len * scr_bytes_elem_size(chunk->elem));
   scr_stream_write_done(s, NULL);
 }
+static void scr_stream_stdio_writables_free(void) {
+  for (int i = 0; i < 2; i++) {
+    ScrStream *s = scr_stream_stdio_writables[i];
+    scr_stream_stdio_writables[i] = NULL;
+    if (s) scr_stream_release(s);
+  }
+}
 ScrStream *scr_stream_stdio_writable(double fd) {
-  ScrClosure *write = scr_closure_new((void *)scr_stream_stdio_write, 1);
-  write->caps[0] = scr_box_new(SCR_BOX_F64);
-  scr_box_set_f64(write->caps[0], fd);
-  return scr_stream_new_writable(-1, true, true, write, scr_stream_stdio_write,
-      NULL, NULL, NULL, NULL);
+  int i = fd == 2 ? 1 : 0;
+  if (scr_stream_stdio_writables[i] == NULL) {
+    if (scr_stream_stdio_writables[1 - i] == NULL) scr_atexit(scr_stream_stdio_writables_free);
+    ScrClosure *write = scr_closure_new((void *)scr_stream_stdio_write, 1);
+    write->caps[0] = scr_box_new(SCR_BOX_F64);
+    scr_box_set_f64(write->caps[0], i ? 2 : 1);
+    scr_stream_stdio_writables[i] = scr_stream_new_writable(-1, true, true, write,
+        scr_stream_stdio_write, NULL, NULL, NULL, NULL);
+  }
+  return scr_stream_retain(scr_stream_stdio_writables[i]);
+}
+
+/* readable.pipe() into a Writable slot under the general representation
+ * (the readable.pipeGeneral libCall): Node never ends process.stdout or
+ * process.stderr from a pipe, so their converted Writable keeps `end` off. */
+ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end) {
+  if (dst == scr_stream_stdio_writables[0] || dst == scr_stream_stdio_writables[1]) end = false;
+  return scr_stream_pipe(src, dst, end);
 }
 
 /* A pipeline() stage that arrives as a checked-dynamic value (the

packages/runtime/src/scr_stream.h

@@ -130,6 +130,9 @@ ScrStream *scr_stream_new_writable_dyn(struct ScrDyn *opts);
  * Writable (fd 1 or 2). */
 ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len);
 ScrStream *scr_stream_stdio_writable(double fd);
+/* readable.pipe() under the general representation: a converted
+ * process.stdout / process.stderr destination is never ended (dst +1). */
+ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end);
 /* A boxed native stream out of a checked-dynamic pipeline stage (+1), or
  * NULL with `message` thrown as the not-supported Error. */
 ScrStream *scr_stream_unbox_dyn(const struct ScrDyn *value, const ScrStr *message);

tests/corpus/6468-node-stdio-writable-slot.ts

@@ -0,0 +1,49 @@
+// process.stdout and process.stderr in NodeJS.WritableStream slots of a
+// program that also passes other streams there: every slot holds the same
+// stream object, a pipe through such a slot leaves the process stream
+// open, and Writable's listener methods return the stream for chaining.
+import { Readable, Writable } from "node:stream";
+
+class Sink extends Writable {
+  _write(_chunk: Buffer, _encoding: string, callback: (error?: Error | null) => void): void {
+    callback();
+  }
+}
+
+function same(a: NodeJS.WritableStream, b: NodeJS.WritableStream): boolean {
+  return a === b;
+}
+
+function pipeInto(dst: NodeJS.WritableStream, parts: string[]): Promise<void> {
+  return new Promise((resolve) => {
+    const src = Readable.from(parts);
+    src.on("end", () => resolve());
+    src.pipe(dst);
+  });
+}
+
+function write(dst: NodeJS.WritableStream, text: string): boolean {
+  return dst.write(text);
+}
+
+async function main(): Promise<void> {
+  console.log("stdout same", same(process.stdout, process.stdout));
+  console.log("stderr same", same(process.stderr, process.stderr));
+  console.log("stdout and stderr", same(process.stdout, process.stderr));
+  console.log("sink and stdout", same(new Sink(), process.stdout));
+
+  await pipeInto(process.stdout, ["piped ", "through ", "a slot\n"]);
+  // The pipe did not end process.stdout: writes through a slot still work.
+  console.log("write after pipe", write(process.stdout, "still open\n"));
+  await pipeInto(process.stderr, ["piped to stderr\n"]);
+  write(process.stderr, "stderr still open\n");
+
+  // Writable's listener methods return the stream.
+  const sink = new Sink();
+  await new Promise<void>((resolve) => {
+    sink.on("finish", () => resolve()).end("done");
+  });
+  console.log("sink finished", sink.writableFinished);
+}
+
+void main();

tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 3500 }