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

src/io/PipeWriter.rs

@@ -518,6 +518,8 @@ 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 holds nothing; `fd` is still the caller's to close.
     pub fn start(&mut self, rawfd: Fd, pollable: bool) -> sys::Result<()> {
         let fd = rawfd;
         self.pollable = pollable;
@@ -526,7 +528,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 +541,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 +1041,7 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         );
     }
 
+    /// On `Err` the writer holds nothing; `fd` is still the caller's to close.
     pub fn start(&mut self, fd: Fd, is_pollable: bool) -> sys::Result<()> {
         if !is_pollable {
             self.close();
@@ -1043,7 +1051,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 +1067,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

@@ -59,8 +59,6 @@ impl PollOrFd {
     ) where
         F: FnOnce(*mut c_void),
     {
-        #[cfg(windows)]
-        let _ = close_fd;
         let fd = self.get_fd();
         #[cfg(target_os = "macos")]
         let mut close_async = true;
@@ -100,7 +98,9 @@ impl PollOrFd {
             // TODO: We should make this call compatible using bun.FD
             #[cfg(windows)]
             {
-                crate::closer::Closer::close(fd, bun_sys::windows::libuv::Loop::get());
+                if close_fd {
+                    crate::closer::Closer::close(fd, bun_sys::windows::libuv::Loop::get());
+                }
             }
             #[cfg(not(windows))]
             {
@@ -128,6 +128,11 @@ impl PollOrFd {
     {
         self.close_impl(ctx, on_close_fn, true);
     }
+
+    /// Unregisters and releases the poll; the fd stays open for its owner to close.
+    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,12 @@ 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.
+                // The writer took neither its ref nor write_fd, and the reader never started.
                 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/js_bun_spawn_bindings.rs

@@ -1764,12 +1764,17 @@ fn spawn_maybe_sync<const IS_SYNC: bool>(
         }
     }
 
-    if let Writable::Buffer(buffer) = subprocess.stdin.get() {
-        if let Err(err) = Writable::buffer_writer_mut(buffer).start() {
-            let _ = subprocess.try_kill(subprocess.kill_signal);
-            let _ = global_this.throw_value(err.to_js(global_this));
-            return Err(JsError::Thrown);
-        }
+    let stdin_start_err = match subprocess.stdin.get() {
+        Writable::Buffer(buffer) => Writable::buffer_writer_mut(buffer).start().err(),
+        _ => None,
+    };
+    if let Some(err) = stdin_start_err {
+        // An unstarted writer never reports on_close; a Buffer left here pins the wrapper.
+        #[cfg(not(windows))] // Windows adopts the pipe at create and start() cannot fail there.
+        subprocess.on_close_io(Subprocess::StdioKind::Stdin);
+        let _ = subprocess.try_kill(subprocess.kill_signal);
+        let _ = global_this.throw_value(err.to_js(global_this));
+        return Err(JsError::Thrown);
     }
 
     **should_close_memfd = false;

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,9 +178,14 @@ impl<P: StaticPipeWriterProcess> StaticPipeWriter<P> {
         }
         #[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

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,175 @@ 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) and that the object whose writer failed
+// is still released: with nothing left to close, the writer never reports
+// on_close, so the owner has to retire it on the error path itself, or a
+// buffer stdin keeps its Subprocess wrapper alive as pending activity forever.
+//
+// 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
+// plus how many fds and Subprocess/Terminal wrappers outlive it, relative to a
+// baseline taken just before. Both classes create their prototype (which
+// heapStats counts under the class name) lazily, so the baseline is taken
+// after materializing it. Releases that happen asynchronously (the child's
+// pidfd once its exit is reaped, a wrapper that becomes collectable only then)
+// get a bounded window; whatever is still there when it lapses is reported.
+const FIXTURE = /* js */ `
+const fs = require("node:fs");
+const { heapStats } = require("bun:jsc");
+const kind = process.argv[2];
+const openFds = () => fs.readdirSync("/proc/self/fd").length;
+const wrappers = () => {
+  const counts = heapStats().objectTypeCounts;
+  return (counts.Subprocess ?? 0) + (counts.Terminal ?? 0);
+};
+
+// Parked on globalThis so the baseline keeps counting it: a local that is never
+// read again is not kept alive across the awaits below.
+if (kind === "terminal") {
+  globalThis.anchor = Bun.Terminal.prototype;
+} else {
+  globalThis.anchor = Bun.spawn({ cmd: ["true"], stdin: "ignore", stdout: "ignore", stderr: "ignore" });
+  await globalThis.anchor.exited;
+}
+const fdBaseline = openFds();
+const wrapperBaseline = wrappers();
+
+let error = null;
+try {
+  switch (kind) {
+    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() > fdBaseline || wrappers() > wrapperBaseline) && performance.now() < deadline) {
+  Bun.gc(true);
+  await Bun.sleep(5);
+}
+console.log(JSON.stringify({ error, leakedFds: openFds() - fdBaseline, leakedWrappers: wrappers() - wrapperBaseline }));
+`;
+
+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) }, leakedFds: 0, leakedWrappers: 0 },
+        stderr: "",
+        exitCode: 0,
+      });
+    });
+
+    test.concurrent(
+      "Bun.spawn with a buffer stdin closes the stdin pipe exactly once and releases the Subprocess",
+      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" },
+            leakedFds: 0,
+            leakedWrappers: 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" }, leakedFds: 0, leakedWrappers: 0 },
+        stderr: "",
+        exitCode: 0,
+      });
+    });
+  },
+);