io(posix): bound the streaming read loop so always-ready sources yield to the event loop

src/io/PipeReader.rs

@@ -453,7 +453,7 @@ impl PosixBufferedReader {
 
     // Exists for consistently with Windows.
     pub fn has_pending_read(&self) -> bool {
-        matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_registered())
+        matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_watching())
     }
 
     pub fn watch(&mut self) {
@@ -872,32 +872,29 @@ impl PosixBufferedReader {
                                 return;
                             }
 
-                            // Keep reading as much as we can
                             if (stack_buffer_len - head_start) < stack_buffer_cutoff {
-                                // `&& !received_hup` mirrors the
-                                // after-inner-loop flush below (line ~855).
-                                // Without it, a peer close (HUP) with >cutoff
-                                // bytes still buffered makes a parent that
-                                // returns `false` on `.eof` (e.g. shell
-                                // `PipeReader::on_read_chunk`) early-return
-                                // here with data left in the kernel and no
-                                // `register_poll`/`done()` → 90s hang in
-                                // shell-blocking-pipe.test.ts.
-                                // Once HUP is set the kernel
-                                // returns the remaining bytes then 0, so
-                                // draining to `bytes_read == 0` is bounded.
-                                if !parent.vtable.on_read_chunk(
+                                let keep_going = parent.vtable.on_read_chunk(
                                     &event_loop.pipe_read_buffer_mut()[..head_start],
                                     if received_hup {
                                         ReadState::Eof
                                     } else {
                                         ReadState::Progress
                                     },
-                                ) && !received_hup
-                                {
-                                    return;
+                                );
+                                // HUP drains to `bytes_read == 0` even if the
+                                // consumer said stop (shell-blocking-pipe).
+                                // A non-pollable File consumer that wants
+                                // more keeps draining to EOF; there is no
+                                // poll to re-arm and the consumer is
+                                // push-driven (FileResponseStream).
+                                if received_hup || (keep_going && file_type == FileType::File) {
+                                    head_start = 0;
+                                    continue;
+                                }
+                                if keep_going {
+                                    parent.register_poll();
                                 }
-                                head_start = 0;
+                                return;
                             }
                         }
                         sys::Result::Err(err) => {
@@ -948,8 +945,9 @@ impl PosixBufferedReader {
                     }
                 }
 
-                if !parent.vtable.is_streaming_enabled() {
-                    break;
+                if file_type != FileType::File {
+                    parent.register_poll();
+                    return;
                 }
             }
         } else if parent._buffer.capacity() == 0 && parent._offset == 0 {

src/runtime/webcore/FileReader.rs

@@ -375,7 +375,7 @@ impl FileReader {
                                 self.reader()
                                     .flags
                                     .set(WindowsFlags::NONBLOCKING, opened.nonblocking);
-                                self.reader().flags.set(WindowsFlags::POLLABLE, pollable);
+                                let _ = pollable;
                             }
                         }
                     }
@@ -563,22 +563,6 @@ impl FileReader {
         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,12 @@ impl FileReader {
             }
         }
 
-        // For pipes, we have to keep pulling or the other process will block.
         // 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());
+        ) && (self.buffered.get().len() + reader_buffer_len < self.highwater_mark);
         close_if_needed!();
         ret
     }

test/js/bun/util/bun-file.test.ts

@@ -1,6 +1,6 @@
-import { expect, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import fsPromises from "fs/promises";
-import { bunEnv, bunExe, tempDirWithFiles } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, isPosix, tempDirWithFiles } from "harness";
 import { join } from "path";
 
 test("delete() and stat() should work with unicode paths", async () => {
@@ -155,3 +155,75 @@ test("Bun.file().json() with UTF-8 BOM does not free an interior pointer", async
   });
   expect(exitCode).toBe(0);
 });
+
+// Before the fix the pollable read loop spun preadv2(RWF_NOWAIT) forever on
+// the JS thread for /dev/urandom and /dev/zero (they never EAGAIN and never
+// EOF), wedging the event loop and growing RSS without bound. Verify that a
+// single read() resolves with a bounded chunk, that a timer scheduled across
+// the read still fires, and that RSS stays flat while the stream sits idle.
+//
+// Sequential on purpose: on a regressed build each child grows RSS ~1 GB/s, so
+// running them concurrently risks OOMing the fail-before step.
+describe.skipIf(!isPosix)("Bun.file(<infinite chardev>).stream() yields to the event loop", () => {
+  // Generous on debug/ASAN (subprocess startup dominates); the release lane
+  // keeps the tight 4 s cap so a regressed build is killed quickly.
+  const hangGuard = isASAN || isDebug ? 20_000 : 4_000;
+
+  for (const [label, source] of [
+    ["Bun.file(dev).stream() on /dev/urandom", `Bun.file("/dev/urandom").stream()`],
+    ["new Response(Bun.file(dev)).body on /dev/zero", `new Response(Bun.file("/dev/zero")).body`],
+  ] as const) {
+    test(
+      label,
+      async () => {
+        await using proc = Bun.spawn({
+          cmd: [
+            bunExe(),
+            "-e",
+            `
+            const rss0 = process.memoryUsage.rss();
+            let tickedAfterRead = false;
+            setTimeout(() => { tickedAfterRead = true; }, 1).unref();
+            const reader = (${source}).getReader();
+            const first = await reader.read();
+            await new Promise(r => setTimeout(r, 10));
+            const second = await reader.read();
+            const rssGrowthMB = (process.memoryUsage.rss() - rss0) / 1024 / 1024;
+            await reader.cancel();
+            process.stdout.write(JSON.stringify({
+              firstLen: first.value?.length ?? -1,
+              secondLen: second.value?.length ?? -1,
+              tickedAfterRead,
+              rssGrowthMB,
+            }));
+          `,
+          ],
+          env: bunEnv,
+          stdout: "pipe",
+          stderr: "pipe",
+          signal: AbortSignal.timeout(hangGuard),
+        });
+        const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+        expect({ stderr, exitCode, signalCode: proc.signalCode }).toEqual({
+          stderr: "",
+          exitCode: 0,
+          signalCode: null,
+        });
+        const out = JSON.parse(stdout);
+        expect(out.tickedAfterRead).toBe(true);
+        expect(out.firstLen).toBeGreaterThan(0);
+        expect(out.firstLen).toBeLessThanOrEqual(1024 * 1024);
+        expect(out.secondLen).toBeGreaterThan(0);
+        expect(out.secondLen).toBeLessThanOrEqual(1024 * 1024);
+        expect(out.rssGrowthMB).toBeLessThan(isASAN || isDebug ? 256 : 128);
+      },
+      hangGuard + 5_000,
+    );
+  }
+
+  test("Bun.file('/dev/null').stream() EOFs immediately", async () => {
+    const r = await Bun.file("/dev/null").stream().getReader().read();
+    expect(r).toEqual({ done: true, value: undefined });
+  });
+});