diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index e073265eb2dd..3a2fb05ac441 100644 --- a/src/io/PipeReader.rs +++ b/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 { diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index da85a4654f4e..298c9c217a21 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/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 } diff --git a/test/js/bun/util/bun-file.test.ts b/test/js/bun/util/bun-file.test.ts index 03e5b1a61fb2..e201e9866a8e 100644 --- a/test/js/bun/util/bun-file.test.ts +++ b/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().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 }); + }); +});