From 02fc8ab13c512df7224be4a1ce6f73f9f604dab2 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Tue, 20 Oct 2026 08:54:18 -0500 Subject: [PATCH] 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. --- .../ambient/scriptc-node-fallback.d.ts | 8 +-- .../compiler/src/backend/llvm/lib-shared.ts | 2 + .../src/frontend/lowering/lower-stream.ts | 14 +++++- packages/compiler/src/ir/builtin-effects.ts | 1 + .../compiler/src/ir/builtin-signatures.ts | 5 +- packages/compiler/src/ir/validate.ts | 4 +- .../6468-node-stdio-writable-slot.ts.json | 6 +++ packages/runtime/src/scr_stream.c | 36 +++++++++++--- packages/runtime/src/scr_stream.h | 3 ++ tests/corpus/6468-node-stdio-writable-slot.ts | 49 +++++++++++++++++++ .../6468-node-stdio-writable-slot.ts.json | 1 + 11 files changed, 114 insertions(+), 15 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json create mode 100644 tests/corpus/6468-node-stdio-writable-slot.ts create mode 100644 tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index 3885e0fef..4e39d9e42 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/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; diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 4602a2bdd..d38379bf0 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -1106,6 +1106,7 @@ export const LIB_FN_SYMS: Record = { "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([ "readable.opReduce", "readable.resume", "readable.pipe", + "readable.pipeGeneral", "readable.unpipe", "writable.write", "writable.writeStr", diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 51c86fe39..88ba1f5b8 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/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, diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index 1c390bbe4..11a292a2e 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -326,6 +326,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "stream.onDyn", "readable.resume", "readable.pipe", + "readable.pipeGeneral", "readable.unpipe", "writable.write", "writable.writeStr", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 597076cb1..82ecf461d 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/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 }, diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 63078ae4a..742787d98 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/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; } diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json new file mode 100644 index 000000000..baf428dee --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/6468-node-stdio-writable-slot.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 1c5e339a0..7c7b2279c 100644 --- a/packages/runtime/src/scr_stream.c +++ b/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 diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index 63ed0c28a..69da3dc0e 100644 --- a/packages/runtime/src/scr_stream.h +++ b/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); diff --git a/tests/corpus/6468-node-stdio-writable-slot.ts b/tests/corpus/6468-node-stdio-writable-slot.ts new file mode 100644 index 000000000..a326344d1 --- /dev/null +++ b/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 { + 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 { + 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((resolve) => { + sink.on("finish", () => resolve()).end("done"); + }); + console.log("sink finished", sink.writableFinished); +} + +void main(); diff --git a/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json b/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json new file mode 100644 index 000000000..3fafdfc4b --- /dev/null +++ b/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json @@ -0,0 +1 @@ +{ "plain": 3000, "san": 3500 }