From 82dcdb809bf8f024ee185db635a978c8e7ecfc0b Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 24 Oct 2026 15:47:26 +0000 Subject: [PATCH 1/3] process.stdin: leave the ref of the paused stdin source for its next reader disown() ended in source.updateRef(false). The next reader of Bun.stdin.stream() resumes the source, but the poll that it arms did not keep the process alive. A console iterator, a Bun.stdin.stream() reader or a Readable.fromWeb() that started after process.stdin.pause(), rl.close() or process.stdin.destroy() read nothing, and the process exited with code 0, unless a top-level await kept it alive. disown() now ends in source.updateRef(!forceUnref). A paused source has no armed poll, so the ref keeps nothing alive until a reader resumes the source. The release of that reader drops the ref. --- src/js/builtins/ProcessObjectInternals.ts | 3 +- test/js/node/process/process-stdin.test.ts | 278 +++++++++++++++++++++ 2 files changed, 280 insertions(+), 1 deletion(-) diff --git a/src/js/builtins/ProcessObjectInternals.ts b/src/js/builtins/ProcessObjectInternals.ts index f4310688b8cb..8c1fa9ad3fad 100644 --- a/src/js/builtins/ProcessObjectInternals.ts +++ b/src/js/builtins/ProcessObjectInternals.ts @@ -151,7 +151,8 @@ export function getStdinStream( reader = undefined; $debug("released reader"); } - source?.updateRef?.(false); + // A paused source keeps nothing alive; its ref is for the next reader of Bun.stdin.stream(). + source?.updateRef?.(!forceUnref); } const ReadStream = isTTY ? require("node:tty").ReadStream : require("node:fs").ReadStream; diff --git a/test/js/node/process/process-stdin.test.ts b/test/js/node/process/process-stdin.test.ts index b1eb1ae830ab..7887cdffb08a 100644 --- a/test/js/node/process/process-stdin.test.ts +++ b/test/js/node/process/process-stdin.test.ts @@ -516,6 +516,284 @@ describe.each([ }); }); +// The children above have a top-level await, which keeps them alive. These do not: the next reader has to +// keep the process alive until stdin ends, and it has to stop when it lets go of stdin too. +describe("the next reader of stdin after process.stdin lets go of it", () => { + const doors = { + "pause()": { first: "", seen: [], door: `process.stdin.pause();` }, + "rl.close()": { + first: "one\n", + seen: ["rl:one"], + door: ` + const rl = require("node:readline").createInterface({ input: process.stdin }); + got.push("rl:" + (await new Promise(resolve => rl.once("line", resolve)))); + rl.close(); + `, + }, + "destroy()": { first: "", seen: [], door: `process.stdin.destroy();` }, + "a break out of for await (process.stdin)": { + first: "one\n", + seen: ["iter:one"], + door: ` + for await (const chunk of process.stdin) { + got.push("iter:" + chunk.toString().trim()); + break; + } + `, + }, + }; + + // `all` reads stdin to its end. `one` reads one chunk and lets go. + const consumers = { + "the console iterator": { + all: `for await (const line of console) got.push(line);`, + one: `for await (const line of console) { got.push(line); break; }`, + }, + "a Bun.stdin.stream() reader": { + all: ` + const reader = Bun.stdin.stream().getReader(); + let text = ""; + for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) { + text += Buffer.from(chunk.value).toString(); + } + got.push(...text.split("\\n")); + `, + one: ` + const reader = Bun.stdin.stream().getReader(); + got.push(Buffer.from((await reader.read()).value).toString().trim()); + reader.releaseLock(); + `, + }, + "Readable.fromWeb(Bun.stdin.stream())": { + all: ` + let text = ""; + for await (const chunk of require("node:stream").Readable.fromWeb(Bun.stdin.stream())) text += chunk; + got.push(...text.split("\\n")); + `, + one: ` + const readable = require("node:stream").Readable.fromWeb(Bun.stdin.stream()); + got.push(String(await new Promise(resolve => readable.once("data", resolve))).trim()); + readable.destroy(); + `, + }, + }; + + const child = (door: string, consumer: string) => ` + const got = []; + (async () => { + ${door} + // process.stdin lets go of the stream a tick after 'pause' or 'close'. + await new Promise(resolve => setImmediate(resolve)); + // An unref'd timer fires only while something else keeps the process alive. + setTimeout(() => console.log("READY"), 50).unref(); + ${consumer} + console.log(JSON.stringify(got)); + })(); + `; + + // Each step waits for `marker` on the child's stdout (if any), then writes `input`. + // With `end: false` stdin stays open, so the child has to exit by itself. + // With `shellPipe` the child's stdin is an anonymous pipe that `cat` writes to, not a socket. + async function run( + script: string, + steps: [marker: string | null, input: string][], + { end = true, shellPipe = false } = {}, + ) { + await using proc = Bun.spawn({ + cmd: shellPipe ? ["sh", "-c", 'cat | "$0" -e "$1"', bunExe(), script] : [bunExe(), "-e", script], + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + env: bunEnv, + }); + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let stdout = ""; + for (const [marker, input] of steps) { + while (marker && !stdout.includes(marker)) { + const { value, done } = await reader.read(); + if (done) throw new Error(`stdout ended before ${marker}; stdout=${JSON.stringify(stdout)}`); + stdout += decoder.decode(value, { stream: true }); + } + proc.stdin.write(input); + proc.stdin.flush(); + } + if (end) await proc.stdin.end(); + for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) { + stdout += decoder.decode(chunk.value, { stream: true }); + } + const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]); + return { stdout, stderr, exitCode, signalCode: proc.signalCode }; + } + + describe.each(Object.entries(doors))("%s", (_, { first, seen, door }) => { + const steps: [string | null, string][] = first ? [[null, first]] : []; + + test.concurrent.each(Object.entries(consumers))("%s reads the rest of stdin", async (_, { all }) => { + expect(await run(child(door, all), [...steps, ["READY\n", "two\nthree"]])).toEqual({ + stdout: "READY\n" + JSON.stringify([...seen, "two", "three"]) + "\n", + stderr: "", + exitCode: 0, + signalCode: null, + }); + }); + + test.concurrent.each(Object.entries(consumers))("%s lets the process exit after one chunk", async (_, { one }) => { + expect(await run(child(door, one), [...steps, ["READY\n", "two\n"]], { end: false })).toEqual({ + stdout: "READY\n" + JSON.stringify([...seen, "two"]) + "\n", + stderr: "", + exitCode: 0, + signalCode: null, + }); + }); + + test.concurrent("the process exits when nothing reads stdin again", async () => { + const script = ` + const got = []; + (async () => { + ${door} + console.log(JSON.stringify(got)); + })(); + `; + expect(await run(script, steps, { end: false })).toEqual({ + stdout: JSON.stringify(seen) + "\n", + stderr: "", + exitCode: 0, + signalCode: null, + }); + }); + }); + + test.skipIf(isWindows).concurrent("the console iterator reads the rest of stdin over a shell pipe", async () => { + const { first, seen, door } = doors["rl.close()"]; + const steps: [string | null, string][] = [ + [null, first], + ["READY\n", "two\nthree"], + ]; + expect(await run(child(door, consumers["the console iterator"].all), steps, { shellPipe: true })).toEqual({ + stdout: "READY\n" + JSON.stringify([...seen, "two", "three"]) + "\n", + stderr: "", + exitCode: 0, + signalCode: null, + }); + }); + + test.concurrent("after process.stdin.unref(), the next reader does not keep the process alive", async () => { + const script = ` + (async () => { + await new Promise(resolve => + process.stdin.once("data", () => { + process.stdin.unref(); + process.stdin.pause(); + resolve(); + }), + ); + await new Promise(resolve => setImmediate(resolve)); + const reader = Bun.stdin.stream().getReader(); + // Stands in for the other work of a program that does not wait for stdin. + const work = setInterval(() => {}, 60_000); + console.log("READY"); + console.log(JSON.stringify(Buffer.from((await reader.read()).value).toString())); + clearInterval(work); + reader.read(); + })(); + `; + expect( + await run( + script, + [ + [null, "one\n"], + ["READY\n", "two"], + ], + { end: false }, + ), + ).toEqual({ stdout: 'READY\n"two"\n', stderr: "", exitCode: 0, signalCode: null }); + }); + + test.concurrent("process.stdin takes stdin back after the next reader releases it", async () => { + const script = ` + const got = []; + (async () => { + process.stdin.once("data", chunk => { + got.push("data:" + chunk.toString().trim()); + process.stdin.pause(); + }); + await new Promise(resolve => process.stdin.once("pause", resolve)); + await new Promise(resolve => setImmediate(resolve)); + const reader = Bun.stdin.stream().getReader(); + setTimeout(() => console.log("READY"), 50).unref(); + got.push("web:" + Buffer.from((await reader.read()).value).toString().trim()); + reader.releaseLock(); + process.stdin.on("data", chunk => got.push("data:" + chunk.toString().trim())); + process.stdin.on("end", () => console.log(JSON.stringify(got))); + process.stdin.resume(); + console.log("AGAIN"); + })(); + `; + expect( + await run(script, [ + [null, "one\n"], + ["READY\n", "two\n"], + ["AGAIN\n", "three\n"], + ]), + ).toEqual({ + stdout: "READY\nAGAIN\n" + JSON.stringify(["data:one", "web:two", "data:three"]) + "\n", + stderr: "", + exitCode: 0, + signalCode: null, + }); + }); + + // readline puts the terminal in raw mode for the prompt, and close() restores it and pauses stdin. + test.skipIf(isWindows).concurrent("the console iterator after a readline prompt on a terminal", async () => { + const script = ` + const got = []; + (async () => { + const rl = require("node:readline").createInterface({ input: process.stdin, output: process.stdout }); + got.push("rl:" + (await new Promise(resolve => rl.question("name? ", resolve)))); + rl.close(); + await new Promise(resolve => setImmediate(resolve)); + setTimeout(() => console.log("READY"), 50).unref(); + for await (const line of console) { + got.push(line); + if (got.length === 3) break; + } + console.log("GOT " + JSON.stringify(got)); + })(); + `; + const decoder = new TextDecoder(); + let output = ""; + let waiting: { marker: string; resolve: () => void } | undefined; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + terminal: { + data(_terminal, chunk) { + output += decoder.decode(chunk, { stream: true }); + if (waiting && output.includes(waiting.marker)) waiting.resolve(); + }, + }, + }); + await using terminal = proc.terminal!; + function waitFor(marker: string) { + return new Promise((resolve, reject) => { + waiting = { marker, resolve }; + proc.exited.then(code => + reject(new Error(`child exited with ${code} before ${marker}; output=${JSON.stringify(output)}`)), + ); + if (output.includes(marker)) resolve(); + }); + } + await waitFor("name? "); + terminal.write("one\r"); + await waitFor("READY"); + terminal.write("two\nthree\n"); + await waitFor("GOT "); + expect(output).toContain('GOT ["rl:one","two","three"]'); + expect(await proc.exited).toBe(0); + }); +}); + // The native FileReader source over a pollable pipe used to drain the fd to // EAGAIN regardless of JS demand, so an idle consumer still ingested the whole // pipe into an internal buffer. The kernel pipe buffer filling up is the From 0156904ad9701677637fa347be05388fadf7f907 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 24 Oct 2026 16:28:16 +0000 Subject: [PATCH 2/3] test: write to the shell pipe of the stdin test from bun, not from cat busybox cat copies with sendfile(). A sendfile() that waits for input on a socket keeps the output pipe locked, so the child blocks in read() on bytes that are already in the pipe. The row timed out on Alpine. --- test/js/node/process/process-stdin.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/js/node/process/process-stdin.test.ts b/test/js/node/process/process-stdin.test.ts index 7887cdffb08a..49816ce9045e 100644 --- a/test/js/node/process/process-stdin.test.ts +++ b/test/js/node/process/process-stdin.test.ts @@ -593,14 +593,17 @@ describe("the next reader of stdin after process.stdin lets go of it", () => { // Each step waits for `marker` on the child's stdout (if any), then writes `input`. // With `end: false` stdin stays open, so the child has to exit by itself. - // With `shellPipe` the child's stdin is an anonymous pipe that `cat` writes to, not a socket. + // With `shellPipe` the child's stdin is an anonymous pipe, not a socket. The writer is not `cat`: + // busybox `cat` uses sendfile(), which locks the pipe while it waits for input and stalls the reader. async function run( script: string, steps: [marker: string | null, input: string][], { end = true, shellPipe = false } = {}, ) { await using proc = Bun.spawn({ - cmd: shellPipe ? ["sh", "-c", 'cat | "$0" -e "$1"', bunExe(), script] : [bunExe(), "-e", script], + cmd: shellPipe + ? ["sh", "-c", '"$0" -e "$2" | "$0" -e "$1"', bunExe(), script, "process.stdin.pipe(process.stdout)"] + : [bunExe(), "-e", script], stdin: "pipe", stdout: "pipe", stderr: "pipe", From c93732375262583672f1d89a90c35a3ab49fdf09 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 24 Oct 2026 16:42:54 +0000 Subject: [PATCH 3/3] test: drain stderr of the stdin test children from the start run() read stderr only after stdout ended, and not at all when stdout ended before a marker. A child that failed early gave no reason, and a child that filled the stderr pipe could not finish. --- test/js/node/process/process-stdin.test.ts | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/test/js/node/process/process-stdin.test.ts b/test/js/node/process/process-stdin.test.ts index 49816ce9045e..253e6eb4fb61 100644 --- a/test/js/node/process/process-stdin.test.ts +++ b/test/js/node/process/process-stdin.test.ts @@ -609,13 +609,18 @@ describe("the next reader of stdin after process.stdin lets go of it", () => { stderr: "pipe", env: bunEnv, }); + // Drained from the start, so a child that fails before its marker shows why. + const stderr = proc.stderr.text(); const reader = proc.stdout.getReader(); const decoder = new TextDecoder(); let stdout = ""; for (const [marker, input] of steps) { while (marker && !stdout.includes(marker)) { const { value, done } = await reader.read(); - if (done) throw new Error(`stdout ended before ${marker}; stdout=${JSON.stringify(stdout)}`); + if (done) { + const output = JSON.stringify({ stdout, stderr: await stderr }); + throw new Error(`stdout ended before ${JSON.stringify(marker)}: ${output}`); + } stdout += decoder.decode(value, { stream: true }); } proc.stdin.write(input); @@ -625,8 +630,7 @@ describe("the next reader of stdin after process.stdin lets go of it", () => { for (let chunk = await reader.read(); !chunk.done; chunk = await reader.read()) { stdout += decoder.decode(chunk.value, { stream: true }); } - const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]); - return { stdout, stderr, exitCode, signalCode: proc.signalCode }; + return { stdout, stderr: await stderr, exitCode: await proc.exited, signalCode: proc.signalCode }; } describe.each(Object.entries(doors))("%s", (_, { first, seen, door }) => {