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,37 @@ impl ReadFile {
         }
     }
 
+    /// `wait_for_readable` for named pipes, on this thread; `false` if the wait itself failed.
+    #[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
+            }
+        }
+    }
+
+    /// A named pipe's whole read, as its own pool task: `JobContext::run` holds a
+    /// VM borrow and VM teardown waits for borrows, so waiting for a writer must
+    /// not happen inside it. Waiting before the first read also keeps a FIFO
+    /// with no writer yet from reading as empty.
+    #[cfg(target_os = "macos")]
+    fn read_named_pipe_task(task: *mut WorkPoolTask) {
+        // SAFETY: only reached via `WorkPoolTask::callback` with `task` =
+        // `&mut self.task` (intrusive) scheduled by `run_async_with_fd`;
+        // recover parent.
+        let this = unsafe { &mut *ReadFile::from_task_ptr(task) };
+        if this.block_until_readable() {
+            this.do_read_loop();
+        } else {
+            this.on_finish();
+        }
+    }
+
     /// 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 +718,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 +795,16 @@ 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 {
+                self.task = WorkPoolTask {
+                    node: Default::default(),
+                    callback: Self::read_named_pipe_task,
+                };
+                WorkPool::schedule(&raw mut self.task);
+                return;
+            }
+
             if bun_core::is_readable(fd) == bun_core::Pollable::NotReady {
                 self.wait_for_readable();
                 return;
@@ -855,6 +906,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,15 @@ 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): no FD_SETSIZE limit; a set is ceil(nfds/32) 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 +7619,41 @@ pub fn kevent(
     }
 }
 
+/// Blocks in `select(2)` until `fd` is readable or at EOF; retries on EINTR.
+///
+/// For named pipes: XNU's kqueue filter for a FIFO (`filt_vnode_common`, which
+/// `poll(2)` uses too) fires only while bytes are buffered, never for the last
+/// writer closing; `fifo_select` consults the FIFO's socket, which reports that
+/// close (and not a writer that has yet to connect). `pipe(2)` pipes report `EV_EOF`.
+#[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"))]