Bun.file(fifo).bytes(): wait for a named pipe's EOF with select(2) on macOS

src/runtime/webcore/blob/read_file.rs

@@ -285,6 +285,9 @@ pub struct ReadFile {
     pub(crate) io_request: io::Request,
     #[cfg(not(windows))]
     pub(crate) could_block: bool,
+    /// FIFO vnode (not a `pipe(2)` pipe); see `bun_sys::block_until_readable`.
+    #[cfg(target_os = "macos")]
+    pub(crate) is_named_pipe: bool,
     pub(crate) close_after_io: bool,
     pub(crate) state: AtomicU8, // ClosingState
 }
@@ -380,6 +383,8 @@ impl ReadFile {
                 scheduled: false,
             },
             could_block: false,
+            #[cfg(target_os = "macos")]
+            is_named_pipe: false,
             close_after_io: false,
             state: AtomicU8::new(ClosingState::Running as u8),
         };
@@ -465,6 +470,21 @@ impl ReadFile {
         }
     }
 
+    /// `wait_for_readable` for named pipes: waits on this (pool) thread.
+    /// Returns `false` if the wait failed, with the error recorded for `then()`.
+    #[cfg(target_os = "macos")]
+    fn block_until_readable(&mut self) -> bool {
+        bloblog!("ReadFile.blockUntilReadable");
+        match bun_sys::block_until_readable(self.opened_fd) {
+            Ok(()) => true,
+            Err(err) => {
+                self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
+                self.system_error = Some(err.to_system_error().into());
+                false
+            }
+        }
+    }
+
     /// Pick the read target: `buffer`'s spare capacity if it is at least as
     /// large as `stack_buffer`, otherwise `stack_buffer`; capped by
     /// `max_length - read_off`. Returns `(use_stack, target)` so the caller
@@ -682,6 +702,11 @@ impl ReadFile {
         }
 
         self.could_block = !bun_sys::is_regular_file(stat.st_mode as _);
+        #[cfg(target_os = "macos")]
+        {
+            // pipe(2) pipes are S_IFIFO with st_dev == 0 (XNU pipe_stat).
+            self.is_named_pipe = bun_sys::S::ISFIFO(stat.st_mode as _) && stat.st_dev != 0;
+        }
         self.total_size =
             SizeType::try_from((stat.st_size as i64).max(0).min(MAX_SIZE as i64)).unwrap();
 
@@ -754,6 +779,17 @@ impl ReadFile {
         // If we immediately call read(), it will block until stdin is
         // readable.
         if self.could_block {
+            #[cfg(target_os = "macos")]
+            if self.is_named_pipe {
+                // Before the first read too: with no writer yet, read() says EOF.
+                if self.block_until_readable() {
+                    self.do_read_loop();
+                } else {
+                    self.on_finish();
+                }
+                return;
+            }
+
             if bun_core::is_readable(fd) == bun_core::Pollable::NotReady {
                 self.wait_for_readable();
                 return;
@@ -855,6 +891,13 @@ impl ReadFile {
                         // call. We already know it's done.
                         && !self.read_eof)
                     {
+                        #[cfg(target_os = "macos")]
+                        if self.is_named_pipe {
+                            if self.block_until_readable() {
+                                continue;
+                            }
+                            break;
+                        }
                         if self.could_block
                         // If we received EOF, we can skip the poll() system
                         // call. We already know it's done.

src/sys/lib.rs

@@ -1411,14 +1411,16 @@ impl Tag {
     #[cfg(not(windows))]
     pub(crate) const setrlimit: Tag = Tag(106);
     pub const clone3: Tag = Tag(107);
+    #[cfg(target_os = "macos")]
+    pub(crate) const select: Tag = Tag(108);
     // `inotify_init1`/`inotify_add_watch` fold under the generic `.watch`
     // tag; `INotifyWatcher.rs` spells it `.inotify`. Alias to `.watch`
     // so the JS-facing `err.syscall == "watch"` string stays node-compatible.
 
     /// The tag name — spelling is frozen (JS-facing
     /// `err.syscall` string; node-compat code matches on it).
     pub fn name(self) -> &'static str {
-        const NAMES: [&str; 108] = [
+        const NAMES: [&str; 109] = [
             "TODO",
             "dup",
             "access",
@@ -1528,6 +1530,7 @@ impl Tag {
             "getrlimit",
             "setrlimit",
             "clone3",
+            "select",
         ];
         NAMES.get(self.0 as usize).copied().unwrap_or("unknown")
     }
@@ -1708,6 +1711,16 @@ mod nocancel {
         ) -> isize;
         #[link_name = "poll$NOCANCEL"]
         pub(crate) fn poll(fds: *mut libc::pollfd, nfds: libc::nfds_t, timeout: c_int) -> c_int;
+        // `_DARWIN_UNLIMITED_SELECT` select(2): the bare syscall, so no
+        // FD_SETSIZE check, and a set is just ceil(nfds / 32) 32-bit words.
+        #[link_name = "select$DARWIN_EXTSN$NOCANCEL"]
+        pub(crate) fn select(
+            nfds: c_int,
+            readfds: *mut u32,
+            writefds: *mut u32,
+            errorfds: *mut u32,
+            timeout: *mut libc::timeval,
+        ) -> c_int;
         // Remaining `$NOCANCEL` variants Bun links against.
         // safe: by-value `c_int` fd; bad fd → -1/EBADF, no UB.
         #[link_name = "close$NOCANCEL"]
@@ -7607,6 +7620,44 @@ pub fn kevent(
     }
 }
 
+/// Blocks in `select(2)` until `fd` is readable or at EOF; retries on EINTR.
+///
+/// For named pipes. XNU hooks a FIFO's `EVFILT_READ` (and `poll(2)`, which it
+/// implements with kqueue) to the vnode, where it only fires while bytes are
+/// buffered, and `fifo_close` posts nothing, so neither ever reports the last
+/// writer closing. `select(2)` goes through `fifo_select` to the FIFO's socket,
+/// whose readability includes `SS_CANTRCVMORE`: set by that close, clear while
+/// no writer has connected yet. `pipe(2)` pipes have their own filter with
+/// `EV_EOF` and do not need this.
+#[cfg(target_os = "macos")]
+pub fn block_until_readable(fd: Fd) -> Maybe<()> {
+    debug_assert!(fd.is_valid());
+    let index = fd.native() as usize;
+    let word = index / 32;
+    let mut read_set = vec![0u32; word + 1];
+    loop {
+        read_set.fill(0);
+        read_set[word] = 1 << (index % 32);
+        // SAFETY: `read_set` holds the ceil(nfds / 32) words the kernel reads
+        // and writes back for `nfds = fd + 1`; the write and error sets and the
+        // timeout (wait indefinitely) may be null.
+        let rc = unsafe {
+            nocancel::select(
+                fd.native() + 1,
+                read_set.as_mut_ptr(),
+                core::ptr::null_mut(),
+                core::ptr::null_mut(),
+                core::ptr::null_mut(),
+            )
+        };
+        match get_errno(rc) {
+            E::SUCCESS => return Ok(()),
+            E::EINTR => continue,
+            e => return Err(Error::from_code(e, Tag::select).with_fd(fd)),
+        }
+    }
+}
+
 /// `clonefile` — macOS-only CoW copy. On non-Darwin returns ENOTSUP so
 /// callers can fall back to `copy_file`.
 #[cfg(not(target_os = "macos"))]

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

@@ -1,5 +1,8 @@
 import { describe, expect, it } from "bun:test";
-import { tempDir } from "harness";
+import { bunEnv, bunExe, isWindows, tempDir } from "harness";
+import { mkfifo } from "mkfifo";
+import { randomBytes } from "node:crypto";
+import { closeSync, constants, openSync, writeSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
 
@@ -54,3 +57,147 @@ describe("Bun.file read-loop target selection", () => {
     expect(Bun.hash(buf)).toBe(Bun.hash(bytes.subarray(start, end)));
   });
 });
+
+// Whole-file reads of a named pipe. Every one of these ends with the reader
+// having drained the pipe and then learning that the last writer closed; on
+// macOS that EOF is invisible to kqueue and poll(2) (they only see buffered
+// bytes on a FIFO), so the reader has to wait for it differently than it does
+// for a pipe(2) pipe, and each of these used to leave the child blocked
+// forever there.
+describe.skipIf(isWindows)("reading a named pipe to EOF", () => {
+  function readFifoInChild(script: string, fifo: string, stdin: number | "ignore" = "ignore") {
+    return Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: { ...bunEnv, FIFO: fifo },
+      stdin,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+  }
+
+  // When a FIFO end gets closed is what these tests are about, so each one is
+  // closed explicitly at the right moment; `using` only covers the failure paths.
+  function openFd(file: string, flags: number | string) {
+    let fd = openSync(file, flags);
+    return {
+      get fd() {
+        return fd;
+      },
+      close() {
+        if (fd !== -1) closeSync(fd);
+        fd = -1;
+      },
+      [Symbol.dispose]() {
+        this.close();
+      },
+    };
+  }
+
+  // The child must already have the FIFO open for reading before a writer can
+  // connect to it: a non-blocking open for writing fails with ENXIO until then.
+  async function openWriterOnceChildIsReading(fifo: string, child: Bun.Subprocess) {
+    while (true) {
+      try {
+        return openFd(fifo, constants.O_WRONLY | constants.O_NONBLOCK);
+      } catch (err: any) {
+        if (err.code !== "ENXIO") throw err;
+      }
+      if (child.exitCode !== null || child.signalCode !== null) {
+        throw new Error(`child exited (${child.exitCode ?? child.signalCode}) without opening the FIFO`);
+      }
+      await Bun.sleep(5);
+    }
+  }
+
+  it.concurrent("bytes() collects a payload that arrives in pieces and ends when the writer closes", async () => {
+    const payload = randomBytes(256 * 1024);
+    using dir = tempDir("bun-file-read-fifo", {});
+    const fifo = path.join(String(dir), "in.fifo");
+    mkfifo(fifo);
+    // The write end can only be opened, and written to without EPIPE, while
+    // some reader has the FIFO open; `holder` is that reader until the child
+    // has opened its own. It never reads, so every byte goes to the child.
+    using holder = openFd(fifo, constants.O_RDONLY | constants.O_NONBLOCK);
+    using writer = openFd(fifo, "w");
+    await using proc = readFifoInChild(
+      `const bytes = await Bun.file(process.env.FIFO).bytes(); process.stdout.write(bytes.length + " " + Bun.hash(bytes));`,
+      fifo,
+    );
+    const stderr = proc.stderr.text();
+    // The write end is blocking, so this write only completes as the child
+    // drains the pipe, and closing it afterwards is what ends the child's
+    // read. The child cannot exit before that unless it failed; dropping
+    // `holder` then leaves the pipe without readers, so the blocked write
+    // fails with EPIPE instead of waiting forever.
+    const childDied = proc.exited.then(async exitCode => {
+      holder.close();
+      throw new Error(`child exited with ${exitCode} before the payload was written: ${await stderr}`);
+    });
+    const written = await Promise.race([Bun.write(Bun.file(writer.fd), payload), childDied]);
+    writer.close();
+    const [stdout, stderrText, exitCode] = await Promise.all([proc.stdout.text(), stderr, proc.exited]);
+
+    expect({ written, stdout, stderr: stderrText }).toEqual({
+      written: payload.length,
+      stdout: `${payload.length} ${Bun.hash(payload)}`,
+      stderr: "",
+    });
+    expect(exitCode).toBe(0);
+  });
+
+  it.concurrent("text() waits for a writer that connects after the read started", async () => {
+    using dir = tempDir("bun-file-read-fifo-late-writer", {});
+    const fifo = path.join(String(dir), "late.fifo");
+    mkfifo(fifo);
+
+    await using proc = readFifoInChild(`process.stdout.write(await Bun.file(process.env.FIFO).text());`, fifo);
+    using writer = await openWriterOnceChildIsReading(fifo, proc);
+    writeSync(writer.fd, "written after the reader opened\n");
+    writer.close();
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({ stdout: "written after the reader opened\n", stderr: "" });
+    expect(exitCode).toBe(0);
+  });
+
+  it.concurrent("text() resolves empty when the writer connects and closes without writing", async () => {
+    using dir = tempDir("bun-file-read-fifo-empty", {});
+    const fifo = path.join(String(dir), "empty.fifo");
+    mkfifo(fifo);
+
+    await using proc = readFifoInChild(
+      `const text = await Bun.file(process.env.FIFO).text(); process.stdout.write(JSON.stringify(text));`,
+      fifo,
+    );
+    using writer = await openWriterOnceChildIsReading(fifo, proc);
+    writer.close();
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({ stdout: '""', stderr: "" });
+    expect(exitCode).toBe(0);
+  });
+
+  it.concurrent("Bun.stdin.text() reads a FIFO inherited as stdin to EOF", async () => {
+    using dir = tempDir("bun-file-read-fifo-stdin", {});
+    const fifo = path.join(String(dir), "stdin.fifo");
+    mkfifo(fifo);
+
+    // Same dance as above: a reader has to exist before the write end can be
+    // opened; here that reader becomes the child's stdin.
+    using readEnd = openFd(fifo, constants.O_RDONLY | constants.O_NONBLOCK);
+    using writer = openFd(fifo, "w");
+    await using proc = readFifoInChild(
+      `process.stdout.write(JSON.stringify(await Bun.stdin.text()));`,
+      fifo,
+      readEnd.fd,
+    );
+    // The child has its own descriptor for the read end now.
+    readEnd.close();
+    writeSync(writer.fd, "stdin is a named pipe\n");
+    writer.close();
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({ stdout: JSON.stringify("stdin is a named pipe\n"), stderr: "" });
+    expect(exitCode).toBe(0);
+  });
+});