io: close a pipe in PollOrFd before it reports closed (POSIX)

src/io/PipeReader.rs

@@ -214,13 +214,20 @@ bitflags::bitflags! {
         const KEEP_ALIVE               = 1 << 10; // default true
         /// A read failed with a non-retry errno. Set before the bytes read ahead of the failure are delivered, so a pull from inside that delivery cannot read the fd past the error. Never cleared: `start()`, `unpause()` and `from()` keep it, and only a reader from `init()` reads again.
         const READ_FAILED              = 1 << 11;
+        /// fstat says that the fd is not a FIFO and not a socket. Only for an fd that the user names (`Bun.file()`): it can be a file, or a device whose close(2) blocks, and fstat cannot tell a pty from a serial line.
+        const NOT_A_PIPE               = 1 << 13;
     }
 }
 
 impl PosixFlags {
     pub(crate) const fn new() -> Self {
         Self::from_bits_truncate(PosixFlags::CLOSE_HANDLE.bits() | PosixFlags::KEEP_ALIVE.bits())
     }
+
+    /// Whether the fd is a pipe, a socket or the pty of `Bun.Terminal`, which `PollOrFd::close_as` closes on the calling thread. A reader knows this before it has a poll.
+    pub(crate) const fn is_pipe(self) -> bool {
+        !self.intersects(PosixFlags::MEMFD.union(PosixFlags::NOT_A_PIPE))
+    }
 }
 
 impl PosixBufferedReader {
@@ -341,12 +348,23 @@ impl PosixBufferedReader {
             debug_assert!(!self.flags.contains(PosixFlags::CLOSED_WITHOUT_REPORTING));
             self.flags.insert(PosixFlags::CLOSED_WITHOUT_REPORTING);
             if self.flags.contains(PosixFlags::CLOSE_HANDLE) {
-                let owner = std::ptr::from_mut(self).cast::<c_void>();
-                self.handle.close(Some(owner), None::<fn(*mut c_void)>);
+                Self::close_fd(&mut self.handle, self.flags, None, None::<fn(*mut c_void)>);
             }
         }
     }
 
+    /// Every close of the fd of a reader. Not `PollOrFd::close`: the poll of a reader does not say what its fd is. Borrows `handle` alone, because `on_close_fn` can free the parent that embeds the reader.
+    fn close_fd<F>(
+        handle: &mut PollOrFd,
+        flags: PosixFlags,
+        ctx: Option<*mut c_void>,
+        on_close_fn: Option<F>,
+    ) where
+        F: FnOnce(*mut c_void),
+    {
+        handle.close_as(ctx, on_close_fn, true, flags.is_pipe());
+    }
+
     pub fn get_fd(&self) -> Fd {
         self.handle.get_fd()
     }
@@ -387,10 +405,10 @@ impl PosixBufferedReader {
                 // The handle is closed after the read regardless of result.
                 // `self.handle` owns the fd;
                 // borrow a non-owning `File` view so the temporary doesn't
-                // close it on drop (handle.close() below does).
+                // close it on drop (`close_fd` below does).
                 let result = sys::File::borrow(&fd)
                     .read_to_end_with_array_list(&mut self._buffer, sys::SizeHint::UnknownSize);
-                self.handle.close(None, None::<fn(*mut c_void)>);
+                Self::close_fd(&mut self.handle, self.flags, None, None::<fn(*mut c_void)>);
                 if let Err(err) = result {
                     bun_core::debug_warn!("error reading from memfd\n{}", err);
                     return self.buffer();
@@ -446,7 +464,9 @@ impl PosixBufferedReader {
         // runs without a receiver borrow.
         unsafe {
             if (*this).flags.contains(PosixFlags::CLOSE_HANDLE) {
-                (*this).handle.close(
+                Self::close_fd(
+                    &mut (*this).handle,
+                    (*this).flags,
                     Some(this.cast::<c_void>()),
                     // SAFETY: ctx is the live reader raw pointer passed above.
                     Some(|ctx: *mut c_void| Self::done(ctx.cast::<PosixBufferedReader>())),
@@ -554,7 +574,7 @@ impl PosixBufferedReader {
         if !is_pollable {
             self.buffer().clear();
             self.flags.remove(PosixFlags::IS_DONE);
-            self.handle.close(None, None::<fn(*mut c_void)>);
+            Self::close_fd(&mut self.handle, self.flags, None, None::<fn(*mut c_void)>);
             self.handle = PollOrFd::Fd(fd);
             return sys::Result::Ok(());
         }

src/io/lib.rs

@@ -1881,6 +1881,11 @@ impl FilePollRef {
         self.inner().owner = owner;
     }
     #[inline]
+    #[cfg(not(windows))]
+    pub(crate) fn owner_tag(self) -> PollTag {
+        self.inner().owner.tag()
+    }
+    #[inline]
     pub(crate) fn deinit_force_unregister(self) {
         self.inner().deinit_force_unregister();
     }

src/io/pipes.rs

@@ -4,8 +4,8 @@ use bun_sys::Fd;
 #[cfg(not(windows))]
 use bun_sys::FdExt;
 
-#[cfg(target_os = "macos")]
-use crate::FilePollFlag;
+#[cfg(not(windows))]
+use crate::{FilePollFlag, PollTag};
 use crate::{FilePollRef, Owner};
 
 pub enum PollOrFd {
@@ -58,35 +58,37 @@ impl PollOrFd {
         close_fd: bool,
     ) where
         F: FnOnce(*mut c_void),
+    {
+        #[cfg(not(windows))]
+        let pipe = match self {
+            PollOrFd::Poll(poll) => {
+                // Only the poll of a writer says what its fd is: a pipe, a socket or the pty of `Bun.Terminal`, or else the tty of a `FileSink`.
+                debug_assert!(!close_fd || poll.owner_tag() != PollTag::BufferedReader);
+                !poll.has_flag(FilePollFlag::Tty)
+            }
+            _ => false,
+        };
+        #[cfg(windows)]
+        let pipe = false;
+        self.close_as(ctx, on_close_fn, close_fd, pipe);
+    }
+
+    /// `close_impl` for a caller that knows whether the fd is a pipe, a socket or the pty of `Bun.Terminal`. The other end waits for the close(2) of one (a child for the end of its stdin), so it runs here, before `on_close_fn`. The close(2) of any other fd can block, so the work pool runs it later.
+    pub fn close_as<F>(
+        &mut self,
+        ctx: Option<*mut c_void>,
+        on_close_fn: Option<F>,
+        close_fd: bool,
+        pipe: bool,
+    ) where
+        F: FnOnce(*mut c_void),
     {
         let fd = self.get_fd();
-        #[cfg(target_os = "macos")]
-        let mut close_async = true;
-        #[cfg(all(not(target_os = "macos"), not(windows)))]
-        let close_async = true;
         if matches!(self, PollOrFd::Poll(_)) {
-            // workaround kqueue bug.
-            // 1) non-blocking FIFO
-            // 2) open for writing only = fd 2, nonblock
-            // 3) open for reading only = fd 3, nonblock
-            // 4) write(3, "something") = 9
-            // 5) read(2, buf, 9) = 9
-            // 6) read(2, buf, 9) = -1 (EAGAIN)
-            // 7) ON ANOTHER THREAD: close(3) = 0,
-            // 8) kevent(2, EVFILT_READ, EV_ADD | EV_ENABLE | EV_DISPATCH, 0, 0, 0) = 0
-            // 9) ??? No more events for fd 2
             // Take ownership of the Box before
             // calling deinit_force_unregister, then leave self = Closed.
             let old = core::mem::replace(self, PollOrFd::Closed);
             if let PollOrFd::Poll(poll) = old {
-                #[cfg(target_os = "macos")]
-                {
-                    if poll.has_flag(FilePollFlag::PollWritable)
-                        && poll.has_flag(FilePollFlag::Nonblocking)
-                    {
-                        close_async = false;
-                    }
-                }
                 // Consumes the underlying allocation.
                 poll.deinit_force_unregister();
             }
@@ -98,18 +100,22 @@ impl PollOrFd {
             // TODO: We should make this call compatible using bun.FD
             #[cfg(windows)]
             {
+                let _ = pipe;
                 if close_fd {
                     crate::closer::Closer::close(fd, bun_sys::windows::libuv::Loop::get());
                 }
             }
             #[cfg(not(windows))]
             {
-                if close_async && close_fd {
+                if close_fd && pipe {
+                    debug_assert!(!fstat_is(fd, bun_sys::S::ISREG));
+                    // Not on another thread: kqueue then delivers no more events to the reader of a nonblocking FIFO.
+                    fd.close();
+                } else if close_fd {
+                    debug_assert!(!fstat_is(fd, |mode| {
+                        bun_sys::S::ISFIFO(mode) || bun_sys::S::ISSOCK(mode)
+                    }));
                     crate::closer::Closer::close(fd, ());
-                } else {
-                    if close_fd {
-                        let _ = fd.close_allowing_bad_file_descriptor(None);
-                    }
                 }
             }
             if let Some(f) = on_close_fn {
@@ -135,6 +141,12 @@ impl PollOrFd {
     }
 }
 
+/// For the debug assertions of `close_as`. False when fstat fails.
+#[cfg(not(windows))]
+fn fstat_is(fd: Fd, is: fn(bun_sys::Mode) -> bool) -> bool {
+    bun_sys::fstat(fd).is_ok_and(|stat| is(stat.st_mode as bun_sys::Mode))
+}
+
 // Sunk to `bun_io` so `FilePoll::file_type()` needs no aio→io edge; re-export
 // keeps the historical `bun_io::FileType` / `bun_io::pipes::FileType` paths.
 pub use crate::posix_event_loop::FileType;

src/runtime/server/FileResponseStream.rs

@@ -646,10 +646,27 @@ bun_io::impl_buffered_reader_parent! {
 impl Drop for FileResponseStream {
     fn drop(&mut self) {
         bun_output::scoped_log!(FileResponseStream, "deinit");
+        let auto_close = self.auto_close.get();
         // `start()` cleared CLOSE_HANDLE, so the reader's `Drop` doesn't unregister the poll.
         #[cfg(unix)]
-        self.reader_mut().handle.close_without_closing_fd();
-        if self.auto_close.get() {
+        {
+            let reader = self.reader_mut();
+            let fd = reader.get_fd();
+            if fd != Fd::INVALID {
+                // `FileType::Pipe` is a character device too, so fstat says whether the other end waits for this close.
+                let pipe = auto_close
+                    && reader.flags.contains(ReaderFlags::POLLABLE)
+                    && sys::fstat(fd).is_ok_and(|stat| {
+                        let mode = stat.st_mode as sys::Mode;
+                        sys::S::ISFIFO(mode) || sys::S::ISSOCK(mode)
+                    });
+                reader
+                    .handle
+                    .close_as(None, None::<fn(*mut c_void)>, auto_close, pipe);
+                return;
+            }
+        }
+        if auto_close {
             #[cfg(windows)]
             Closer::close(self.fd.get(), bun_sys::windows::libuv::Loop::get());
             #[cfg(not(windows))]

src/runtime/webcore/FileReader.rs

@@ -100,6 +100,9 @@ pub(crate) struct OpenedFileBlob {
     pub(crate) nonblocking: bool,
     #[cfg(not(windows))]
     pub(crate) file_type: FileType,
+    /// Not a FIFO and not a socket. `file_type` is a pipe for every device that can be polled.
+    #[cfg(not(windows))]
+    pub(crate) not_a_pipe: bool,
 }
 
 impl Default for OpenedFileBlob {
@@ -110,6 +113,8 @@ impl Default for OpenedFileBlob {
             nonblocking: true,
             #[cfg(not(windows))]
             file_type: FileType::File,
+            #[cfg(not(windows))]
+            not_a_pipe: false,
         }
     }
 }
@@ -224,6 +229,7 @@ impl Lazy {
             } else {
                 FileType::File
             };
+            this.not_a_pipe = this.file_type == FileType::File;
 
             // pretend it's a non-blocking pipe if it's a TTY
             if is_nonblocking && this.file_type != FileType::Socket {
@@ -338,6 +344,9 @@ impl FileReader {
                                     .flags
                                     .set(PosixFlags::NONBLOCKING, opened.nonblocking);
                                 self.reader().flags.set(PosixFlags::POLLABLE, pollable);
+                                self.reader()
+                                    .flags
+                                    .set(PosixFlags::NOT_A_PIPE, opened.not_a_pipe);
                             }
                             #[cfg(windows)]
                             {

test/js/bun/shell/shell-pipe-read-fault.test.ts

@@ -248,10 +248,8 @@ console.log(JSON.stringify({ exitCode: r.exitCode }));
 // has to register its poll (the stdout/stderr readers register theirs right
 // after). Runs the command several times and lists the parent's fds that were
 // not open before: a spawn that fails part way through setup must not leave a
-// stdio end open. An fd held by a registered poll is closed on the thread pool,
-// so the list is polled until it drains (bounded). A command that never
-// finishes shows up as a test timeout. `argv[2]` selects the single-command or
-// the pipeline form.
+// stdio end open. A command that never finishes shows up as a test timeout.
+// `argv[2]` selects the single-command or the pipeline form.
 const STDIN_BLOB_FIXTURE = /* js */ `
 import { $ } from "bun";
 import { readdirSync, readlinkSync } from "node:fs";
@@ -265,12 +263,7 @@ await run();
 const before = fds();
 let last;
 for (let i = 0; i < 5; i++) last = await run();
-const deadline = Date.now() + 2000;
-let leaked = [...fds()].filter(fd => !before.has(fd));
-while (leaked.length > 0 && Date.now() < deadline) {
-  await Bun.sleep(10);
-  leaked = [...fds()].filter(fd => !before.has(fd));
-}
+const leaked = [...fds()].filter(fd => !before.has(fd));
 console.log(JSON.stringify({ exitCode: last.exitCode, stderr: last.stderr.toString().trim(), leaked }));
 `;
 

test/js/bun/spawn/spawn-pipe-read-error-leak.test.ts

@@ -99,19 +99,10 @@ async function once() {
 
   Bun.gc(true);
 
-  // The fd close goes through bun.Async.Closer which schedules close() on
-  // a jsc.WorkPool thread; Bun.sleep(0) doesn't synchronize with that. Poll
-  // a few times so a briefly-starved close thread doesn't flag a false
-  // positive (same pattern as spawn-stdin-pipe-fd-leak.test.ts).
+  // The reader closed its fd when it handled the failed read.
   let leaked = 0;
-  for (let i = 0; i < 20; i++) {
-    const final = snapshotFds();
-    leaked = 0;
-    for (const [fd, link] of final) {
-      if (before.get(fd) !== link) leaked++;
-    }
-    if (leaked === 0) break;
-    await Bun.sleep(20);
+  for (const [fd, link] of snapshotFds()) {
+    if (before.get(fd) !== link) leaked++;
   }
   return leaked;
 }

test/js/bun/spawn/spawn-pipe-start-error.test.ts

@@ -224,8 +224,7 @@ if (kind !== "terminal") {
   globalThis.anchor.push(child);
   await child.exited;
 }
-// The first spawnSync of a process creates the loop it waits on, which stays. Without pipes: theirs are closed on
-// the work pool, some time after the call.
+// The first spawnSync of a process creates the loop it waits on, which stays.
 const quiet = { cmd: ["true"], stdin: "ignore", stdout: "ignore", stderr: "ignore" };
 if (kind.includes("unwatchable") || kind.startsWith("sync-") || kind.endsWith("-running")) Bun.spawnSync(quiet);
 const fdBaseline = openFds();

test/js/bun/spawn/spawn-stdin-pipe-fd-leak.test.ts

@@ -44,20 +44,17 @@ test.skipIf(!isPosix)("stdin: 'pipe' fd is closed on child exit without reading
   }
   await Promise.all(children.map(p => p.exited));
 
-  // fd closes go through bun.Async.Closer on POSIX; give the close thread
-  // a moment — but do NOT invoke GC, since the bug was that cleanup only
-  // happened via Subprocess finalization.
-  for (let i = 0; i < 20 && countOpenFds() - baseline > N / 4; i++) await Bun.sleep(20);
-
+  // Do NOT invoke GC: the bug was that cleanup only happened via Subprocess
+  // finalization.
   const afterExit = countOpenFds() - baseline;
 
   // Keep `children` alive across the measurement so GC finalization cannot
   // be what closed the fds.
   expect(children.length).toBe(N);
 
   // Without the fix every stdin pipe (one per child) is still open here,
-  // so `afterExit` ≈ N. Allow slack for a few async closes still in flight.
-  expect(afterExit).toBeLessThan(N / 4);
+  // so `afterExit` ≈ N.
+  expect(afterExit).toBe(0);
 });
 
 test("reading .stdin does not leak a native FileSink per spawn", async () => {

test/js/bun/spawn/spawn-stdin-readable-stream-edge-cases.test.ts

@@ -555,15 +555,7 @@ describe("spawn stdin ReadableStream edge cases", () => {
         }
         return pids.length;
       }
-      // Some are closed on the work pool, or by a collection.
-      async function leakedFds() {
-        const deadline = performance.now() + 2000;
-        while (openFds() > baseline && performance.now() < deadline) {
-          Bun.gc(true);
-          await Bun.sleep(5);
-        }
-        return openFds() - baseline;
-      }
+      const leakedFds = () => openFds() - baseline;
     `;
 
     async function run(code: string) {

test/js/bun/spawn/spawn-streaming-stdin.test.ts

@@ -63,13 +63,7 @@ test("spawn can write to stdin multiple chunks", async () => {
     remaining -= concurrency;
   }
 
-  // Pipe fds are closed on a background thread after their streams finish, so
-  // the last batch's closes may still be in flight here; wait for them to land.
-  let newMaxFD = getMaxFD();
-  for (let i = 0; i < 100 && newMaxFD !== maxFD; i++) {
-    await Bun.sleep(10);
-    newMaxFD = getMaxFD();
-  }
+  const newMaxFD = getMaxFD();
 
   // assert we didn't leak any file descriptors
   expect(newMaxFD).toBe(maxFD);