stdin: apply highwater backpressure to the pipe FileReader source

src/io/PipeReader.rs

@@ -389,6 +389,13 @@ impl PosixBufferedReader {
     /// embedding `self` (the shell `PipeReader` does exactly that), so the
     /// caller must not touch `self` again after a `false` return.
     pub fn register_poll(&mut self) -> bool {
+        // A pause issued from inside the JS re-entry of the read loop
+        // (on_read_chunk → microtasks → setFlowing(false) → pause()) lands
+        // while the loop's own `register_poll()` is still ahead on the stack;
+        // re-arming here would undo it. `unpause()` + `read()` re-arm.
+        if self.flags.contains(PosixFlags::IS_PAUSED) {
+            return true;
+        }
         // Hoist vtable-derived scalars and
         // normalize self.handle to Poll before taking the single &mut borrow,
         // so no raw-pointer escape is needed.
@@ -503,6 +510,9 @@ impl PosixBufferedReader {
     }
 
     pub fn on_poll(parent: &mut PosixBufferedReader, size_hint: isize, received_hup: bool) {
+        if parent.flags.contains(PosixFlags::IS_PAUSED) {
+            return;
+        }
         let fd = parent.get_fd();
         bun_sys::syslog!("onPoll({}) = {}", fd, size_hint);
 
@@ -662,14 +672,18 @@ impl PosixBufferedReader {
 
                         if streaming {
                             // Stream this chunk and register for next cycle
-                            let _ = parent.vtable.on_read_chunk(
+                            if !parent.vtable.on_read_chunk(
                                 &stack_buffer[..bytes_read],
                                 if received_hup && bytes_read < stack_buffer.len() {
                                     ReadState::Eof
                                 } else {
                                     ReadState::Progress
                                 },
-                            );
+                            ) && !received_hup
+                                && !over_budget
+                            {
+                                return;
+                            }
                         } else {
                             parent
                                 ._buffer

src/runtime/webcore/FileReader.rs

@@ -563,22 +563,6 @@
         true
     }
 
-    #[inline]
-    fn reader_is_pollable(&self) -> bool {
-        #[cfg(unix)]
-        {
-            self.reader()
-                .flags
-                .contains(bun_io::pipe_reader::PosixFlags::POLLABLE)
-        }
-        #[cfg(windows)]
-        {
-            self.reader()
-                .flags
-                .contains(bun_io::pipe_reader::WindowsFlags::POLLABLE)
-        }
-    }
-
     pub fn on_read_chunk(&self, init_buf: &[u8], state: ReadState) -> bool {
         let mut buf = init_buf;
         bun_core::scoped_log!(
@@ -809,14 +793,19 @@
             }
         }
 
-        // For pipes, we have to keep pulling or the other process will block.
+        // No JS read is waiting; the chunk was appended to `self.buffered`.
+        // Stop the read loop once the buffered bytes reach the highwater mark
+        // or the consumer paused (`setFlowing(false)`). A full kernel pipe
+        // buffer blocking the writer is the backpressure signal. `onPull` /
+        // `setFlowing(true)` re-derive demand and restart the reader.
         // SAFETY: see `reader_buffer` decl.
         let reader_buffer_len = unsafe { (*reader_buffer).len() };
-        let ret = !matches!(
-            self.read_inside_on_pull.get(),
-            ReadDuringJSOnPullResult::Temporary(_)
-        ) && !(self.buffered.get().len() + reader_buffer_len >= self.highwater_mark
-            && !self.reader_is_pollable());
+        let ret = self.flowing.get()
+            && !matches!(
+                self.read_inside_on_pull.get(),
+                ReadDuringJSOnPullResult::Temporary(_)
+            )
+            && self.buffered.get().len() + reader_buffer_len < self.highwater_mark;
         close_if_needed!();
         ret
     }
@@ -965,6 +954,16 @@
         self.pending_value.with_mut(|p| p.set(&global, array));
         self.pending_view.set(buffer);
 
+        // The read loop returns without re-arming the one-shot poll when
+        // `on_read_chunk` reports the highwater backstop, and pause() skips
+        // the unregister while the poll is already disarmed, so
+        // `has_pending_read()` (which tracks the registration flag, not the
+        // kernel arm state) can be true with no wakeup scheduled. The
+        // consumer is asking for data here, so make sure the poll is armed.
+        if self.flowing.get() {
+            self.reader().watch();
+        }
+
         bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len);
 
         streams::Result::Pending(self.pending.as_ptr())

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

@@ -1,5 +1,5 @@
-import { expect, test } from "bun:test";
-import { bunEnv, bunExe } from "harness";
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN, isDebug, isWindows } from "harness";
 
 test.concurrent("pipe does the right thing", async () => {
   // Note: Bun.spawnSync uses memfd_create on Linux for pipe, which means we see
@@ -418,3 +418,114 @@ test.concurrent("pause() and resume() churn while data is in flight never destro
   expect(stdout.trim()).toBe(`TOTAL ${20 * 1024}`);
   expect(exitCode).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
+// backpressure signal; these tests feed far more than that and check the
+// child's resident set does not grow to match.
+describe.skipIf(isWindows)("pipe backpressure", () => {
+  const feedMB = 40;
+  // With no backpressure the child buffers the whole feed (Vec growth roughly
+  // doubles that in RSS). With backpressure only the highwater mark plus the
+  // kernel pipe buffer are resident in the child.
+  const maxDeltaMB = isASAN || isDebug ? 24 : 16;
+
+  async function run(consumer: string) {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", consumer],
+      stdin: "pipe",
+      stdout: "pipe",
+      stderr: "pipe",
+      env: bunEnv,
+    });
+    // The child exits under backpressure having accepted only a few KB, so
+    // the queued writes here fail with EPIPE; that is the expected outcome.
+    const chunk = Buffer.alloc(1024 * 1024, 0x78);
+    const ignoreEpipe = (e: any) => {
+      if (e?.code !== "EPIPE") throw e;
+    };
+    for (let i = 0; i < feedMB; i++) {
+      const r = proc.stdin.write(chunk);
+      if (r && typeof (r as any).then === "function") (r as Promise<number>).catch(ignoreEpipe);
+    }
+    Promise.resolve(proc.stdin.flush()).catch(ignoreEpipe);
+    Promise.resolve(proc.stdin.end()).catch(ignoreEpipe);
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    const result = JSON.parse(stdout.trim());
+    expect(exitCode).toBe(0);
+    return result;
+  }
+
+  test.concurrent("Bun.stdin.stream(): a single read does not ingest the whole pipe", async () => {
+    const { first, deltaMB } = await run(`
+      const rd = Bun.stdin.stream().getReader();
+      const c = await rd.read();
+      const base = process.memoryUsage().rss;
+      // Give the event loop time to (incorrectly) drain the pipe. The loop is
+      // native (no JS on the no-pending path), so debug/ASAN overhead is small.
+      await new Promise(r => setTimeout(r, 1500));
+      const deltaMB = Math.round((process.memoryUsage().rss - base) / 1048576);
+      process.stdout.write(JSON.stringify({ first: c.value?.length ?? 0, deltaMB }));
+      process.exit(0);
+    `);
+    expect(first).toBeGreaterThan(0);
+    expect(deltaMB).toBeLessThan(maxDeltaMB);
+  });
+
+  test.concurrent("process.stdin.pause() stops the fd from being read", async () => {
+    const { bytesAfter, deltaMB } = await run(`
+      let bytes = 0, pausedAt = 0;
+      process.stdin.on("data", chunk => {
+        bytes += chunk.length;
+        if (!pausedAt && bytes >= 1 << 20) {
+          pausedAt = bytes;
+          process.stdin.pause();
+          const base = process.memoryUsage().rss;
+          setTimeout(() => {
+            const deltaMB = Math.round((process.memoryUsage().rss - base) / 1048576);
+            process.stdout.write(JSON.stringify({ bytesAtPause: pausedAt, bytesAfter: bytes, deltaMB }));
+            process.exit(0);
+          }, 1500);
+        }
+      });
+    `);
+    expect(bytesAfter).toBeLessThan(feedMB * 1024 * 1024);
+    expect(deltaMB).toBeLessThan(maxDeltaMB);
+  });
+
+  test.concurrent("reading resumes after the highwater backstop", async () => {
+    // Stop reading long enough for the backstop to engage, then drain to EOF
+    // and make sure every byte written by the parent is delivered.
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const rd = Bun.stdin.stream().getReader();
+        await rd.read().then(c => { globalThis.total = c.value?.length ?? 0; });
+        await new Promise(r => setTimeout(r, 200));
+        while (true) {
+          const { value, done } = await rd.read();
+          if (value) total += value.length;
+          if (done) break;
+        }
+        process.stdout.write(String(total));
+        `,
+      ],
+      stdin: "pipe",
+      stdout: "pipe",
+      stderr: "pipe",
+      env: bunEnv,
+    });
+    const chunk = Buffer.alloc(64 * 1024, 0x79);
+    const n = 64;
+    for (let i = 0; i < n; i++) proc.stdin.write(chunk);
+    await proc.stdin.end();
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    expect(stdout).toBe(String(n * chunk.length));
+    expect(exitCode).toBe(0);
+  });
+});