Bun.write: deliver the whole payload to a FIFO instead of a torn prefix

src/runtime/webcore/Blob.rs

@@ -1354,6 +1354,7 @@ impl BlobExt for Blob {
                 mkdirp_if_not_exists,
                 extra_options: options,
                 mode: None,
+                handoff_fd: None,
             },
         )
     }
@@ -4456,6 +4457,9 @@ pub struct WriteFileOptions {
     pub mkdirp_if_not_exists: Option<bool>,
     pub extra_options: Option<JSValue>,
     pub mode: Option<bun_sys::Mode>,
+    /// Destination fd already opened by the sync fast path, for `WriteFile`
+    /// to adopt instead of reopening.
+    pub handoff_fd: Option<Fd>,
 }
 
 /// Write an empty string to a file by truncating it.
@@ -4674,6 +4678,14 @@ pub fn write_file_with_source_destination(
     destination_blob: &mut Blob,
     options: &WriteFileOptions,
 ) -> JsResult<JSValue> {
+    // Close the handoff fd on any branch that doesn't adopt it.
+    let handoff_fd = core::cell::Cell::new(options.handoff_fd);
+    let _close_handoff_fd = scopeguard::guard((), |()| {
+        if let Some(fd) = handoff_fd.get() {
+            let _ = bun_sys::close(fd);
+        }
+    });
+
     let destination_store = destination_blob
         .store
         .get()
@@ -4734,6 +4746,10 @@ pub fn write_file_with_source_destination(
                 options.mkdirp_if_not_exists.unwrap_or(true),
             )
             .expect("unreachable");
+            if let Some(fd) = handoff_fd.take() {
+                // SAFETY: file_copier was just produced by heap::into_raw; sole owner.
+                unsafe { (*file_copier).opened_fd = fd };
+            }
             let task = write_file_mod::WriteFileTask::create_on_js_thread(ctx, file_copier);
             // Defer promise creation until we're just about to schedule the task.
             // `JSPromiseStrong.strong` is private in `bun_jsc`, so use `init` (which
@@ -5040,13 +5056,24 @@ pub fn write_file_internal(
         }
     }
 
+    #[cfg(not(windows))]
+    let handoff_fd = core::cell::Cell::new(None::<Fd>);
+    // Close the fast-path fd on any early return before WriteFile adopts it.
+    #[cfg(not(windows))]
+    let _close_handoff_fd = scopeguard::guard((), |()| {
+        if let Some(fd) = handoff_fd.get() {
+            let _ = bun_sys::close(fd);
+        }
+    });
+
     // If you're doing Bun.write(), try to go fast by writing short input on the main thread.
     // This is a heuristic, but it's a good one.
     //
     // except if you're on Windows. Windows I/O is slower. Let's not even try.
     #[cfg(not(windows))]
     {
         let mut needs_async = false;
+        let mut fast_path_fd: Option<Fd> = None;
         let fast_path_ok = matches!(*path_or_blob, PathOrBlob::Path(_))
             || (matches!(*path_or_blob, PathOrBlob::Blob(ref b)
                 if b.offset.get() == 0 && !b.is_s3()
@@ -5075,13 +5102,15 @@ pub fn write_file_internal(
                             &pathlike,
                             str.get(),
                             &mut needs_async,
+                            &mut fast_path_fd,
                         )
                     } else {
                         write_string_to_file_fast::<false>(
                             global_this,
                             &pathlike,
                             str.get(),
                             &mut needs_async,
+                            &mut fast_path_fd,
                         )
                     };
                     if !needs_async {
@@ -5106,13 +5135,15 @@ pub fn write_file_internal(
                             &pathlike,
                             buffer_view.byte_slice(),
                             &mut needs_async,
+                            &mut fast_path_fd,
                         )
                     } else {
                         write_bytes_to_file_fast::<false>(
                             global_this,
                             &pathlike,
                             buffer_view.byte_slice(),
                             &mut needs_async,
+                            &mut fast_path_fd,
                         )
                     };
                     if !needs_async {
@@ -5121,6 +5152,7 @@ pub fn write_file_internal(
                 }
             }
         }
+        handoff_fd.set(fast_path_fd);
     }
 
     // if path_or_blob is a path, convert it into a file blob
@@ -5285,6 +5317,12 @@ pub fn write_file_internal(
     // StoreRef clone+drop keeps the destination store alive across the call.
     let _dest_hold = destination_store;
 
+    #[cfg(not(windows))]
+    let options = WriteFileOptions {
+        handoff_fd: handoff_fd.take(),
+        ..options
+    };
+
     write_file_with_source_destination(
         global_this,
         &mut *source_blob,
@@ -5375,6 +5413,7 @@ pub fn write_file(global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResu
             mkdirp_if_not_exists,
             extra_options: options,
             mode,
+            handoff_fd: None,
         },
     )
 }
@@ -5387,6 +5426,7 @@ fn write_string_to_file_fast<const NEEDS_OPEN: bool>(
     pathlike: &PathOrFileDescriptor,
     str: BunString,
     needs_async: &mut bool,
+    handoff_fd: &mut Option<Fd>,
 ) -> JSValue {
     let fd: Fd = if !NEEDS_OPEN {
         pathlike.fd()
@@ -5413,6 +5453,20 @@ fn write_string_to_file_fast<const NEEDS_OPEN: bool>(
         }
     };
 
+    // Non-regular files (FIFO/socket/chardev) can EAGAIN; route them to the
+    // async WriteFile path before any bytes go out so it can wait for POLLOUT.
+    if !str.is_empty() {
+        if let bun_sys::Result::Ok(st) = bun_sys::fstat(fd) {
+            if !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode) {
+                *needs_async = true;
+                if NEEDS_OPEN {
+                    *handoff_fd = Some(fd);
+                }
+                return JSValue::ZERO;
+            }
+        }
+    }
+
     // Declared before the truncate guard so it drops *after* it (close runs last).
     let _close = NEEDS_OPEN.then(|| bun_sys::CloseOnDrop::new(fd));
 
@@ -5470,7 +5524,8 @@ fn write_bytes_to_file_fast<const NEEDS_OPEN: bool>(
     global_this: &JSGlobalObject,
     pathlike: &PathOrFileDescriptor,
     bytes: &[u8],
-    _needs_async: &mut bool,
+    needs_async: &mut bool,
+    handoff_fd: &mut Option<Fd>,
 ) -> JSValue {
     let fd: Fd = if !NEEDS_OPEN {
         pathlike.fd()
@@ -5490,7 +5545,7 @@ 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::ENOENT {
-                    *_needs_async = true;
+                    *needs_async = true;
                     return JSValue::ZERO;
                 }
                 return JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
@@ -5501,6 +5556,20 @@ fn write_bytes_to_file_fast<const NEEDS_OPEN: bool>(
         }
     };
 
+    // Non-regular files (FIFO/socket/chardev) can EAGAIN; route them to the
+    // async WriteFile path before any bytes go out so it can wait for POLLOUT.
+    if !bytes.is_empty() {
+        if let bun_sys::Result::Ok(st) = bun_sys::fstat(fd) {
+            if !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode) {
+                *needs_async = true;
+                if NEEDS_OPEN {
+                    *handoff_fd = Some(fd);
+                }
+                return JSValue::ZERO;
+            }
+        }
+    }
+
     // TODO: on windows this is always synchronous
 
     let truncate = NEEDS_OPEN || bytes.is_empty();
@@ -5520,7 +5589,7 @@ 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;
+                    *needs_async = true;
                     return JSValue::ZERO;
                 }
                 let err_js = if !NEEDS_OPEN {

src/runtime/webcore/S3Client.rs

@@ -594,6 +594,7 @@ impl S3Client {
                 mkdirp_if_not_exists: Some(false),
                 extra_options: options,
                 mode: None,
+                handoff_fd: None,
             },
         )
     }

src/runtime/webcore/blob/write_file.rs

@@ -333,15 +333,15 @@ impl WriteFile {
         let fd = self.opened_fd;
         debug_assert!(fd != Fd::INVALID);
 
-        // We do not use pwrite() because the file may not be
-        // seekable (such as stdout)
-        //
-        // 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 {
+            // We do not use pwrite() because the file may not be
+            // seekable (such as stdout)
+            //
+            // 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]);
+
             match &result {
                 bun_sys::Result::Ok(res) => {
                     *wrote = *res;
@@ -464,10 +464,11 @@ impl WriteFile {
                 }
             }
 
-            // 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.
+            // Path-opened fds have no cached mode; fstat so a FIFO/socket
+            // reaches wait_for_writable() on EAGAIN.
+            if let bun_sys::Result::Ok(st) = sys::fstat(fd) {
+                break 'brk !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode);
+            }
             false
         };
 

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

@@ -1,6 +1,7 @@
 import { describe, expect, it, test } from "bun:test";
 import fs, { mkdirSync } from "fs";
 import { bunEnv, bunExe, exampleHtml, exampleSite, gcTick, isWindows, tempDir, withoutAggressiveGC } from "harness";
+import { mkfifo } from "mkfifo";
 import path, { join } from "path";
 
 let i = 0;
@@ -727,4 +728,68 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) {
 
     expect(f.name).toBe(filePath);
   });
+
+  // Writing more than the kernel pipe buffer to a FIFO used to deliver only the
+  // first partial write to the reader and then either reject ENXIO (fast path
+  // closed and reopened the fifo) or never settle (async path spun on EAGAIN
+  // with could_block=false). Both must now deliver the full payload.
+  describe.skipIf(isWindows).each([
+    ["string", payload => payload],
+    ["Uint8Array", payload => new TextEncoder().encode(payload)],
+    ["Blob", payload => new Blob([payload])],
+    ["Response", payload => new Response(payload)],
+  ])("Bun.write(fifo, %s) larger than the pipe buffer", (label, toSource) => {
+    // 200003 bytes: well over the 64 KiB Linux pipe buffer (and the 8 KiB
+    // minimum some CI kernels use), with a distinct tail so a torn prefix
+    // is visible.
+    const N = 200003;
+    const payload = Buffer.alloc(N - 3, "x").toString() + "END";
+
+    async function exercise(suffix, writeTo) {
+      using dir = tempDir(`bun-write-fifo-${label}-${suffix}`, {});
+      const fifo = join(String(dir), "f.fifo");
+      mkfifo(fifo);
+
+      // Open the read side synchronously with O_NONBLOCK so it returns
+      // immediately (no writer required) and the writer's O_WRONLY|O_NONBLOCK
+      // open cannot observe ENXIO. Reading through Bun.file(fd).stream() polls
+      // on EAGAIN instead of parking a thread-pool worker, which matters under
+      // describe.concurrent.
+      const readFd = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
+      const drained = (async () => {
+        const chunks = [];
+        for await (const chunk of Bun.file(readFd).stream()) chunks.push(chunk);
+        return Buffer.concat(chunks);
+      })();
+      drained.catch(() => {});
+
+      let got;
+      let written;
+      try {
+        written = await writeTo(fifo, toSource(payload));
+        got = await drained;
+      } finally {
+        fs.closeSync(readFd);
+        await drained.catch(() => {});
+      }
+
+      expect({ bytes: got.length, tail: got.subarray(-3).toString(), written }).toEqual({
+        bytes: N,
+        tail: "END",
+        written: N,
+      });
+    }
+
+    it("delivers every byte to the reader (path dest)", () => exercise("path", (fifo, src) => Bun.write(fifo, src)));
+
+    it("delivers every byte to the reader (O_NONBLOCK fd dest)", () =>
+      exercise("fd", async (fifo, src) => {
+        const wfd = fs.openSync(fifo, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK);
+        try {
+          return await Bun.write(Bun.file(wfd), src);
+        } finally {
+          fs.closeSync(wfd);
+        }
+      }));
+  });
 });