process.stdin: deliver 'end' when pause() inside a 'data' handler races a buffered EOF

src/js/builtins/ProcessObjectInternals.ts

@@ -167,6 +167,7 @@ export function getStdinStream(
 
   let stream_destroyed = false;
   let stream_reachedEof = false;
+  let hasReceivedData = false;
   stream.addListener = stream.on = function (event, listener) {
     // Streams don't generally required to present any data when only
     // `readable` events are present, i.e. `readableFlowing === false`
@@ -215,9 +216,9 @@ export function getStdinStream(
   const originalRead = stream.read;
   stream.read = function (size) {
     const ret = originalRead.$call(this, size);
-    // An explicit read() must acquire the native reader: _read() without one only
-    // records needsInternalReadRefresh, which own() replays. Owning afterwards so a
-    // throwing size never refs stdin; read(0) kicks never own (pause() relies on it).
+    // An explicit sized read() must acquire the native reader when _read() did
+    // not (buffer already satisfied it, or hasReceivedData is false). Owning
+    // after originalRead so a throwing size never refs stdin; read(0) defers.
     if (size !== 0 && reader === undefined && !stream_destroyed) {
       own();
     }
@@ -252,6 +253,7 @@ export function getStdinStream(
 
     try {
       if (value) {
+        hasReceivedData = true;
         stream.push(value);
       } else {
         // EOF. Nothing is left to read, so release the native reader before
@@ -263,6 +265,10 @@ export function getStdinStream(
         // return the buffered < n byte remainder and 'end'/'readableEnded'
         // come from the stream machinery once the buffer drains.
         stream.push(null);
+        // push(null) only marks the state ended; endReadable() runs on the next
+        // read(). Node's onStreamRead issues read(0) here so 'end' fires even
+        // when pause() has stopped flow() and nothing else will call read().
+        stream.read(0);
       }
     } catch (err) {
       if (value) triggerRead.$call(stream, undefined);
@@ -281,7 +287,14 @@ export function getStdinStream(
       needsInternalReadRefresh = true;
     }
   }
-  stream._read = triggerRead;
+  // Node's Socket._read calls readStart(): Readable asking for data re-arms the
+  // native read after pause()'s readStop(), so a queued pipe EOF is delivered
+  // via maybeReadMore() -> _read(). TTY stays disowned so a child can inherit.
+  function triggerReadOwning(size) {
+    if (!reader && hasReceivedData && !isTTY && !stream_reachedEof && !stream_destroyed) own();
+    triggerRead.$call(this, size);
+  }
+  stream._read = triggerReadOwning;
 
   stream.on("resume", () => {
     if (stream.isPaused()) return; // fake resume

test/js/node/process/process-stdin.test.ts

@@ -259,6 +259,62 @@ test.concurrent("'end' is not emitted when the buffer is never drained, and the
   expect(await stdioResult(proc)).toEqual({ stdout: "EXIT 8\n", exitCode: 0 });
 });
 
+test.concurrent("pausing inside a 'data' handler with EOF already buffered still delivers 'end'", async () => {
+  // Producer writes all lines + closes in one go; child pauses on the first chunk.
+  // The EOF queued behind that chunk must still reach push(null) and emit 'end'.
+  const proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const out = [];
+      process.stdin.on("data", d => { out.push("data:" + d.length); process.stdin.pause(); });
+      process.stdin.on("end", () => out.push("end"));
+      process.stdin.resume();
+      process.on("exit", () => console.log(JSON.stringify({ out, readableEnded: process.stdin.readableEnded })));`,
+    ],
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+    env: bunEnv,
+  });
+  proc.stdin.write("a\nb\nc\n");
+  proc.stdin.end();
+
+  expect(await stdioResult(proc)).toEqual({
+    stdout: JSON.stringify({ out: ["data:6", "end"], readableEnded: true }) + "\n",
+    exitCode: 0,
+  });
+});
+
+test.concurrent("readline close() inside a 'line' handler with piped EOF still lets stdin emit 'end'", async () => {
+  // rl.close() pauses process.stdin; the remaining lines still drain, and the
+  // already-buffered pipe EOF must reach 'end' + readableEnded rather than
+  // being parked behind the paused stream.
+  const proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const readline = require("node:readline");
+      const out = [];
+      const rl = readline.createInterface({ input: process.stdin });
+      rl.on("line", l => { out.push("line:" + l); if (out.length === 1) rl.close(); });
+      process.stdin.on("end", () => out.push("end"));
+      process.on("exit", () => console.log(JSON.stringify({ out, readableEnded: process.stdin.readableEnded })));`,
+    ],
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+    env: bunEnv,
+  });
+  proc.stdin.write("a\nb\nc\n");
+  proc.stdin.end();
+
+  expect(await stdioResult(proc)).toEqual({
+    stdout: JSON.stringify({ out: ["line:a", "line:b", "line:c", "end"], readableEnded: true }) + "\n",
+    exitCode: 0,
+  });
+});
+
 test.concurrent("stdin should allow process to exit when paused", async () => {
   const proc = Bun.spawn({
     cmd: [