Bun.write: poll instead of spin when a nonblocking stdout pipe returns EAGAIN

src/runtime/webcore/Blob.rs

@@ -5445,8 +5445,22 @@ fn write_string_to_file_fast<const NEEDS_OPEN: bool>(
                 bun_sys::Result::Err(err) => {
                     truncate.set(false);
                     if err.get_errno() == bun_sys::E::EAGAIN {
-                        *needs_async = true;
-                        return JSValue::ZERO;
+                        if written.get() == 0 {
+                            *needs_async = true;
+                            return JSValue::ZERO;
+                        }
+                        // Part of the payload is already committed; handing the rest to the
+                        // async path would re-send the whole input and duplicate bytes. This
+                        // branch can only be reached on a pollable fd, so block here until
+                        // it drains (equivalent to the blocking write() we would have issued
+                        // had the fd not been flipped O_NONBLOCK behind our back).
+                        let mut pfd = [bun_sys::posix::PollFd {
+                            fd: fd.native(),
+                            events: bun_sys::posix::POLL_OUT,
+                            revents: 0,
+                        }];
+                        let _ = bun_sys::posix::poll(&mut pfd, -1);
+                        continue;
                     }
                     let err_js = if !NEEDS_OPEN {
                         err.to_js(global_this)
@@ -5520,8 +5534,17 @@ fn write_bytes_to_file_fast<const NEEDS_OPEN: bool>(
             bun_sys::Result::Err(err) => {
                 #[cfg(not(windows))]
                 if err.get_errno() == bun_sys::E::EAGAIN {
-                    *_needs_async = true;
-                    return JSValue::ZERO;
+                    if written == 0 {
+                        *_needs_async = true;
+                        return JSValue::ZERO;
+                    }
+                    let mut pfd = [bun_sys::posix::PollFd {
+                        fd: fd.native(),
+                        events: bun_sys::posix::POLL_OUT,
+                        revents: 0,
+                    }];
+                    let _ = bun_sys::posix::poll(&mut pfd, -1);
+                    continue;
                 }
                 let err_js = if !NEEDS_OPEN {
                     err.to_js(global_this)

src/runtime/webcore/blob/write_file.rs

@@ -338,35 +338,25 @@ impl WriteFile {
         //
         // On macOS, it is an error to use pwrite() on a
         // non-seekable file.
-        let result: bun_sys::Result<usize> =
-            sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]);
-
-        loop {
-            match &result {
-                bun_sys::Result::Ok(res) => {
-                    *wrote = *res;
-                    self.total_written += *res;
-                }
-                bun_sys::Result::Err(err) => {
-                    if err.get_errno() == io::RETRY {
-                        if !self.could_block {
-                            // regular files cannot use epoll.
-                            // this is fine on kqueue, but not on epoll.
-                            continue;
-                        }
-                        self.wait_for_writable();
-                        return false;
-                    } else {
-                        self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
-                        self.system_error = Some(err.to_system_error().into());
-                        return false;
-                    }
+        match sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]) {
+            bun_sys::Result::Ok(res) => {
+                *wrote = res;
+                self.total_written += res;
+                true
+            }
+            bun_sys::Result::Err(err) => {
+                if err.get_errno() == io::RETRY {
+                    // write(2) on a regular file never returns EAGAIN, so reaching here means
+                    // the fd is pollable regardless of what `could_block` was initialised to.
+                    self.could_block = true;
+                    self.wait_for_writable();
+                } else {
+                    self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
+                    self.system_error = Some(err.to_system_error().into());
                 }
+                false
             }
-            break;
         }
-
-        true
     }
 
     pub fn then(mut this: Box<WriteFile>, _global: &JSGlobalObject) -> Result<(), JsTerminated> {
@@ -414,15 +404,21 @@ impl WriteFile {
     }
 
     pub fn is_allowed_to_close(&self) -> bool {
-        self.file_blob
+        match self
+            .file_blob
             .store
             .get()
             .as_ref()
             .unwrap()
             .data
             .as_file()
             .pathlike
-            .is_path()
+        {
+            PathOrFileDescriptor::Path(_) => true,
+            // A caller-supplied fd stays owned by the caller unless `run_with_fd` duped it
+            // for polling, in which case `opened_fd` is the private dup and must be closed.
+            PathOrFileDescriptor::Fd(fd) => self.opened_fd != Fd::INVALID && self.opened_fd != fd,
+        }
     }
 
     #[cfg(not(windows))]
@@ -450,26 +446,46 @@ impl WriteFile {
 
         let fd = self.opened_fd;
 
-        self.could_block = 'brk: {
-            if let Some(store) = self.file_blob.store.get().as_ref() {
-                if let blob::store::Data::File(file) = &store.data {
-                    if file.pathlike.is_fd() {
-                        // If seekable was set, then so was mode
-                        if file.seekable.is_some() {
-                            // This is mostly to handle pipes which were passsed to the process somehow
-                            // such as stderr, stdout. Bun.stdin and Bun.stderr will automatically set `mode` for us.
-                            break 'brk !bun_sys::is_regular_file(file.mode);
-                        }
+        let caller_supplied_fd = match self.file_blob.store.get().as_ref() {
+            Some(store) => match &store.data {
+                blob::store::Data::File(file) => match file.pathlike {
+                    PathOrFileDescriptor::Fd(_) => {
+                        // Bun.stdout/Bun.stderr fstat() up front and set `mode` (but not
+                        // `seekable`), so gate on `mode` being populated.
+                        self.could_block = file.mode != 0 && !bun_sys::is_regular_file(file.mode);
+                        true
                     }
+                    PathOrFileDescriptor::Path(_) => {
+                        // We opened the file descriptor with O_NONBLOCK, so we
+                        // shouldn't have to worry about blocking reads/writes
+                        //
+                        // We do not call fstat() because that is very expensive.
+                        self.could_block = false;
+                        false
+                    }
+                },
+                _ => false,
+            },
+            None => false,
+        };
+
+        // A caller-supplied pollable fd (e.g. Bun.stdout backed by a pipe) may be the target
+        // of several concurrent Bun.write() calls. Each WriteFile carries its own `io::Poll`,
+        // and the IO thread's epoll/kqueue keys interest by fd number, so two instances
+        // registering the same fd collide (EEXIST on Linux; last-wins udata on kqueue). Dup it
+        // so this instance owns a private fd number for the same open file description; the dup
+        // is closed in `on_finish` via `is_allowed_to_close`.
+        if self.could_block && caller_supplied_fd {
+            match bun_sys::dup(fd) {
+                bun_sys::Result::Ok(duped) => self.opened_fd = duped,
+                bun_sys::Result::Err(err) => {
+                    self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
+                    self.system_error = Some(err.to_system_error().into());
+                    self.on_finish();
+                    return;
                 }
             }
-
-            // We opened the file descriptor with O_NONBLOCK, so we
-            // shouldn't have to worry about blocking reads/writes
-            //
-            // We do not call fstat() because that is very expensive.
-            false
-        };
+        }
 
         // We have never supported offset in Bun.write().
         // and properly adding support means we need to also support it

test/js/bun/io/bun-write.test.js

@@ -728,3 +728,47 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) {
     expect(f.name).toBe(filePath);
   });
 });
+
+// The POSIX async WriteFile path used to compute `could_block = false` for Bun.stdout (the stdio
+// store sets `mode` but not `seekable`), and its EAGAIN handler re-matched a cached write()
+// result in a tight loop instead of re-issuing the syscall or polling. Once fd 1 was O_NONBLOCK
+// (process.stdout.write's FileSink sets it on the shared open file description via a dup), every
+// Bun.write that overflowed the pipe buffer wedged a thread-pool worker at 100% CPU with the
+// returned promise never settling.
+//
+// Runs outside `describe.concurrent` because on an unfixed build the child pegs every core until
+// the spawn timeout fires, which would starve the concurrent neighbours into their 5s default.
+it.skipIf(isWindows)(
+  "Bun.write(Bun.stdout, ...) to a full nonblocking pipe completes",
+  async () => {
+    // Two payload sizes: below 256 KiB exercises the sync fast path's EAGAIN -> needs_async
+    // fallback; at/above 256 KiB goes straight to the thread-pool WriteFile path.
+    const script = `
+    process.stdout.write("x"); // constructs the fd 1 FileSink, which flips the pipe O_NONBLOCK
+    const small = Buffer.alloc(64 * 1024, 65).toString();
+    const large = Buffer.alloc(256 * 1024, 66).toString();
+    const ps = [];
+    for (let i = 0; i < 32; i++) ps.push(Bun.write(Bun.stdout, small));
+    for (let i = 0; i < 8; i++) ps.push(Bun.write(Bun.stdout, large));
+    const wrote = await Promise.all(ps);
+    process.stderr.write("wrote=" + wrote.reduce((a, b) => a + b, 0));
+  `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+      timeout: 10_000,
+      killSignal: "SIGKILL",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.bytes(), proc.stderr.text(), proc.exited]);
+    const expected = 1 + 32 * 64 * 1024 + 8 * 256 * 1024;
+    expect({ length: stdout.length, stderr, exitCode, signalCode: proc.signalCode }).toEqual({
+      length: expected,
+      stderr: "wrote=" + (expected - 1),
+      exitCode: 0,
+      signalCode: null,
+    });
+  },
+  15_000,
+);