io: leave the fd with the caller when a POSIX pipe writer fails to start

src/io/PipeWriter.rs

@@ -518,6 +518,9 @@ impl<Parent: PosixBufferedWriterParent> PosixBufferedWriter<Parent> {
 
     /// On POSIX a `MovableIfWindowsFd` never transfers ownership, so callers
     /// pass the plain `Fd` (via `MovableIfWindowsFd::get_posix()` when needed).
+    ///
+    /// On `Err` the writer does not hold `fd`: the caller still owns it and is
+    /// the one that closes it.
     pub fn start(&mut self, rawfd: Fd, pollable: bool) -> sys::Result<()> {
         let fd = rawfd;
         self.pollable = pollable;
@@ -526,7 +529,8 @@ impl<Parent: PosixBufferedWriterParent> PosixBufferedWriter<Parent> {
             self.handle = PollOrFd::Fd(fd);
             return sys::Result::Ok(());
         }
-        let poll = match self.get_poll() {
+        let existing_poll = self.get_poll();
+        let poll = match existing_poll {
             Some(p) => p,
             None => {
                 let p = self.create_poll(fd);
@@ -538,6 +542,10 @@ impl<Parent: PosixBufferedWriterParent> PosixBufferedWriter<Parent> {
 
         match poll.register_with_fd(loop_, FilePollKind::Writable, fd) {
             sys::Result::Err(err) => {
+                // A poll from an earlier start() still holds that start's fd.
+                if existing_poll.is_none() {
+                    self.handle.close_without_closing_fd();
+                }
                 return sys::Result::Err(err);
             }
             sys::Result::Ok(()) => {
@@ -1034,6 +1042,8 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         );
     }
 
+    /// On `Err` the writer does not hold `fd`: the caller still owns it and is
+    /// the one that closes it.
     pub fn start(&mut self, fd: Fd, is_pollable: bool) -> sys::Result<()> {
         if !is_pollable {
             self.close();
@@ -1043,7 +1053,8 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
 
         // SAFETY: parent BACKREF set via set_parent; outlives this writer.
         let loop_ = unsafe { Parent::event_loop(self.parent()) };
-        let poll = match self.get_poll() {
+        let existing_poll = self.get_poll();
+        let poll = match existing_poll {
             Some(p) => p,
             None => {
                 let p = FilePollRef::init(
@@ -1058,6 +1069,10 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
 
         match poll.register_with_fd(loop_.loop_(), FilePollKind::Writable, fd) {
             sys::Result::Err(err) => {
+                // A poll from an earlier start() still holds that start's fd.
+                if existing_poll.is_none() {
+                    self.handle.close_without_closing_fd();
+                }
                 return sys::Result::Err(err);
             }
             sys::Result::Ok(()) => {}

src/io/pipes.rs

@@ -128,6 +128,12 @@
     {
         self.close_impl(ctx, on_close_fn, true);
     }
+
+    /// Releases the poll (unregistering it) and leaves the fd open: for owners
+    /// that close the fd themselves.
+    pub fn close_without_closing_fd(&mut self) {
+        self.close_impl(None, None::<fn(*mut c_void)>, false);
+    }
 }
 
 // Sunk to `bun_io` so `FilePoll::file_type()` needs no aio→io edge; re-export

src/runtime/api/bun/Terminal.rs

@@ -499,20 +499,15 @@ impl Terminal {
         {
             sys::Result::Ok(()) => terminal.ref_(),
             sys::Result::Err(_) => {
-                // POSIX: writer.start() may have allocated a poll holding write_fd
-                // before registerWithFd failed; closeInternal → writer.close()
-                // frees the poll and closes write_fd. Windows: writer.start()
-                // failure leaves source==null so writer.close() is a no-op; close
-                // write_fd directly. Pre-set writer_done so onWriterClose's deref
-                // is skipped and the struct isn't freed mid-closeInternal.
+                // A failed writer.start() leaves write_fd with us: close both pty
+                // fds here, closeInternal releases master/slave. The writer holds
+                // nothing, so its close() reports nothing; WRITER_DONE keeps
+                // onWriterClose a no-op either way, as the only ref is dropped below.
                 terminal.update_flags(|f| f.insert(Flags::WRITER_DONE));
                 terminal.read_fd.get().close();
                 terminal.read_fd.set(Fd::INVALID);
-                #[cfg(windows)]
-                {
-                    terminal.write_fd.get().close();
-                    terminal.write_fd.set(Fd::INVALID);
-                }
+                terminal.write_fd.get().close();
+                terminal.write_fd.set(Fd::INVALID);
                 terminal.close_internal();
                 terminal.deref_();
                 return Err(InitError::WriterStartFailed);

src/runtime/api/bun/subprocess/Writable.rs

@@ -297,15 +297,18 @@ impl<'a> Writable<'a> {
         match stdio {
             Stdio::Dup2(_) => panic!("TODO dup2 stdio"),
             Stdio::Pipe | Stdio::ReadableStream(_) => {
+                let fd = result.unwrap();
                 // `create` returns a freshly-boxed non-null pointer.
-                let pipe_nn = NonNull::new(FileSink::create(evtloop, result.unwrap()))
+                let pipe_nn = NonNull::new(FileSink::create(evtloop, fd))
                     .expect("FileSink::create returns non-null");
                 let pipe = Self::pipe_sink_mut(&pipe_nn);
 
-                match pipe.writer.with_mut(|w| w.start(pipe.fd.get(), true)) {
+                match pipe.writer.with_mut(|w| w.start(fd, true)) {
                     bun_sys::Result::Ok(()) => {}
                     bun_sys::Result::Err(_err) => {
                         Self::pipe_release(pipe_nn);
+                        // The writer did not take `fd`; nothing else closes it.
+                        fd.close();
                         if let Stdio::ReadableStream(rs) = stdio {
                             rs.cancel(global);
                         }

src/runtime/shell/IOWriter.rs

@@ -368,12 +368,6 @@ impl IOWriter {
                     s.flags.pollable = false;
                     s.flags.nonblock = false;
                     s.flags.is_socket = false;
-                    if matches!(s.writer.handle, bun_io::pipes::PollOrFd::Poll(_)) {
-                        s.writer
-                            .handle
-                            .close_impl(None, None::<fn(*mut c_void)>, false);
-                    }
-                    s.writer.handle = bun_io::pipes::PollOrFd::Closed;
                     return self.__start();
                 }
                 #[cfg(any(target_os = "linux", target_os = "android"))]
@@ -384,12 +378,6 @@ impl IOWriter {
                         s.flags.pollable = false;
                         s.flags.nonblock = false;
                         s.flags.is_socket = false;
-                        if matches!(s.writer.handle, bun_io::pipes::PollOrFd::Poll(_)) {
-                            s.writer
-                                .handle
-                                .close_impl(None, None::<fn(*mut c_void)>, false);
-                        }
-                        s.writer.handle = bun_io::pipes::PollOrFd::Closed;
                         return self.__start();
                     }
                 }

src/spawn/static_pipe_writer.rs

@@ -178,10 +178,15 @@
         }
         #[cfg(not(windows))]
         {
+            use bun_sys::FdExt as _;
             // On POSIX `StdioResult` is an `Option<Fd>`.
-            match self.writer.start(self.stdio_result.unwrap(), true) {
+            let fd = self.stdio_result.unwrap();
+            match self.writer.start(fd, true) {
                 bun_sys::Result::Err(err) => {
+                    // The writer did not take `fd`; nothing else closes it.
+                    self.stdio_result = None;
+                    fd.close();
                     // start() failed: `started` stays false so no release
                     // site fires — release start()'s `+1` here.
                     // SAFETY: `self` is the live `Self` we ref'd at the top
                     // of `start()`; the caller's `IntrusiveRc` keeps it alive

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

@@ -1,5 +1,6 @@
-import { expect, test } from "bun:test";
-import { bunEnv, bunExe, isDebug, isWindows } from "harness";
+import { afterAll, beforeAll, describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isDebug, isLinux, isWindows, tempDir } from "harness";
+import { join } from "node:path";
 
 // On Windows, when the initial uv_read_start on a subprocess stdout/stderr
 // pipe fails (observed from libuv as UV_EINVAL after a bad FileAccessInformation
@@ -62,3 +63,146 @@ try {
     expect(exitCode).toBe(0);
   },
 );
+
+// POSIX counterpart for the writers. When a pipe writer's start() fails to
+// register its fd with the event loop, the writer no longer holds the fd and
+// the caller that opened it closes it: the subprocess stdin FileSink
+// (Writable::init), the buffered stdin StaticPipeWriter, and Bun.Terminal's
+// pty writer each do so on their own error path. Each fixture provokes that
+// failure and checks that the fd is closed exactly once: an fd left open shows
+// up in /proc/self/fd, a second close of the same number trips the debug
+// build's EBADF assertion on stderr.
+//
+// Bun registers FilePolls through syscall(SYS_epoll_ctl, ...) rather than the
+// epoll_ctl() wrapper, so the shim interposes syscall() and fails every
+// EPOLL_CTL_ADD asking for writability with ENOSPC (what an exhausted
+// fs.epoll.max_user_watches returns). Readable registrations, and uSockets,
+// which uses the wrapper, are unaffected.
+const cc = Bun.which("cc") || Bun.which("gcc") || Bun.which("clang");
+
+const SHIM_C = /* c */ `
+#define _GNU_SOURCE
+#include <dlfcn.h>
+#include <errno.h>
+#include <stdarg.h>
+#include <sys/epoll.h>
+#include <sys/syscall.h>
+
+static long (*real_syscall)(long, ...);
+
+long syscall(long number, ...) {
+  va_list ap;
+  va_start(ap, number);
+  long a1 = va_arg(ap, long), a2 = va_arg(ap, long), a3 = va_arg(ap, long);
+  long a4 = va_arg(ap, long), a5 = va_arg(ap, long), a6 = va_arg(ap, long);
+  va_end(ap);
+  if (number == SYS_epoll_ctl && a2 == EPOLL_CTL_ADD && a4 != 0 &&
+      (((struct epoll_event *)a4)->events & EPOLLOUT)) {
+    errno = ENOSPC;
+    return -1;
+  }
+  if (!real_syscall) real_syscall = (long (*)(long, ...))dlsym(RTLD_NEXT, "syscall");
+  return real_syscall(number, a1, a2, a3, a4, a5, a6);
+}
+`;
+
+// The argument selects what to construct; the report is the error it threw
+// and how many fds are still open afterwards, compared to before. Fds that
+// the error path releases asynchronously (the child's pidfd once its exit is
+// reaped, anything owned by a wrapper that is only released by its finalizer)
+// get a bounded window to go away; a leaked fd is reported once it lapses.
+const FIXTURE = /* js */ `
+const fs = require("node:fs");
+const openFds = () => fs.readdirSync("/proc/self/fd").length;
+const before = openFds();
+let error = null;
+try {
+  switch (process.argv[2]) {
+    case "stdin-pipe":
+      Bun.spawn({ cmd: ["true"], stdin: "pipe", stdout: "ignore", stderr: "ignore" });
+      break;
+    case "stdin-buffer":
+      Bun.spawn({ cmd: ["true"], stdin: Buffer.from("data"), stdout: "ignore", stderr: "ignore" });
+      break;
+    case "terminal":
+      new Bun.Terminal({});
+      break;
+  }
+} catch (e) {
+  error = { code: e.code, message: e.message };
+}
+const deadline = performance.now() + 2000;
+while (openFds() > before && performance.now() < deadline) {
+  Bun.gc(true);
+  await Bun.sleep(5);
+}
+console.log(JSON.stringify({ error, leaked: openFds() - before }));
+`;
+
+describe.skipIf(!isLinux || !cc)(
+  "a pipe writer whose event loop registration fails leaves its fd to the caller",
+  () => {
+    let dir: ReturnType<typeof tempDir>;
+
+    beforeAll(async () => {
+      dir = tempDir("writer-start-error", { "shim.c": SHIM_C, "fixture.js": FIXTURE });
+      await using ccProc = Bun.spawn({
+        cmd: [cc!, "-shared", "-fPIC", "-o", join(String(dir), "shim.so"), join(String(dir), "shim.c"), "-ldl"],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [ccOut, ccErr, ccExit] = await Promise.all([ccProc.stdout.text(), ccProc.stderr.text(), ccProc.exited]);
+      if (ccExit !== 0) throw new Error(`shim compile failed: ${ccErr || ccOut}`);
+    });
+
+    afterAll(() => {
+      dir?.[Symbol.dispose]();
+    });
+
+    async function runFixture(kind: string, env: Record<string, string> = {}) {
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), "fixture.js", kind],
+        cwd: String(dir),
+        env: { ...bunEnv, ...env, LD_PRELOAD: join(String(dir), "shim.so") },
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      let report: unknown = stdout;
+      try {
+        report = JSON.parse(stdout);
+      } catch {}
+      return { report, stderr, exitCode };
+    }
+
+    test.concurrent("Bun.spawn with stdin: 'pipe' closes the stdin pipe exactly once", async () => {
+      expect(await runFixture("stdin-pipe")).toEqual({
+        // The spawn bindings report a failed stdin setup generically, so only
+        // the fact that it threw is pinned down here.
+        report: { error: { message: expect.any(String) }, leaked: 0 },
+        stderr: "",
+        exitCode: 0,
+      });
+    });
+
+    test.concurrent("Bun.spawn with a buffer stdin closes the stdin pipe exactly once", async () => {
+      // On Linux a buffer stdin normally travels through a memfd and never gets
+      // a writer; disabling that takes the pipe writer path every other
+      // platform uses.
+      expect(await runFixture("stdin-buffer", { BUN_FEATURE_FLAG_DISABLE_MEMFD: "1" })).toEqual({
+        report: { error: { code: "ENOSPC", message: "ENOSPC: no space left on device, epoll_ctl" }, leaked: 0 },
+        stderr: "",
+        exitCode: 0,
+      });
+    });
+
+    test.concurrent("new Bun.Terminal() closes the pty fds exactly once", async () => {
+      expect(await runFixture("terminal")).toEqual({
+        report: { error: { message: "Failed to start terminal writer" }, leaked: 0 },
+        stderr: "",
+        exitCode: 0,
+      });
+    });
+  },
+);