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

src/runtime/webcore/Blob.rs

@@ -1354,6 +1354,7 @@
                 mkdirp_if_not_exists,
                 extra_options: options,
                 mode: None,
+                handoff_fd: None,
             },
         )
     }
@@ -4456,6 +4457,10 @@
     pub mkdirp_if_not_exists: Option<bool>,
     pub extra_options: Option<JSValue>,
     pub mode: Option<bun_sys::Mode>,
+    /// An fd the sync fast path already opened for the destination. When set,
+    /// the async `WriteFile` adopts it instead of reopening the path so a
+    /// FIFO's reader never sees an intermediate EOF.
+    pub handoff_fd: Option<Fd>,
 }
 
 /// Write an empty string to a file by truncating it.
@@ -4674,6 +4679,16 @@
     destination_blob: &mut Blob,
     options: &WriteFileOptions,
 ) -> JsResult<JSValue> {
+    // The fast path may have opened the destination and handed the fd over.
+    // Only the posix File←Bytes branch adopts it; any other branch (or an
+    // early error) must not leak 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 +4749,10 @@
                 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 +5059,26 @@
         }
     }
 
+    #[cfg(not(windows))]
+    let handoff_fd = core::cell::Cell::new(None::<Fd>);
+    // If the fast path opened an fd and handed it off, make sure it closes on
+    // any early-return between here and `write_file_with_source_destination`
+    // consuming 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 +5107,15 @@
                             &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 +5140,15 @@
                             &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 +5157,7 @@
                 }
             }
         }
+        handoff_fd.set(fast_path_fd);
     }
 
     // if path_or_blob is a path, convert it into a file blob
@@ -5285,6 +5322,12 @@
     // 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 +5418,7 @@
             mkdirp_if_not_exists,
             extra_options: options,
             mode,
+            handoff_fd: None,
         },
     )
 }
@@ -5387,6 +5431,7 @@
     pathlike: &PathOrFileDescriptor,
     str: BunString,
     needs_async: &mut bool,
+    handoff_fd: &mut Option<Fd>,
 ) -> JSValue {
     let fd: Fd = if !NEEDS_OPEN {
         pathlike.fd()
@@ -5413,6 +5458,20 @@
         }
     };
 
+    // A FIFO/socket/chardev opened O_NONBLOCK will EAGAIN once the pipe
+    // buffer fills. Closing mid-payload would deliver a torn prefix and EOF
+    // to the reader, so hand the open fd to the async WriteFile path instead
+    // and let it drive POLLOUT. Regular files never EAGAIN; keep the fast path.
+    if NEEDS_OPEN {
+        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;
+                *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 +5529,8 @@
     global_this: &JSGlobalObject,
     pathlike: &PathOrFileDescriptor,
     bytes: &[u8],
     _needs_async: &mut bool,
+    handoff_fd: &mut Option<Fd>,
 ) -> JSValue {
     let fd: Fd = if !NEEDS_OPEN {
         pathlike.fd()
@@ -5501,6 +5561,19 @@
         }
     };
 
+    // See write_string_to_file_fast: route non-regular files to the async
+    // path with the fd we just opened, so a FIFO write never closes+reopens
+    // mid-payload.
+    if NEEDS_OPEN {
+        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;
+                *handoff_fd = Some(fd);
+                return JSValue::ZERO;
+            }
+        }
+    }
+
     // TODO: on windows this is always synchronous
 
     let truncate = NEEDS_OPEN || bytes.is_empty();

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,13 @@ 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.
+            // We opened with O_NONBLOCK. For a regular file that flag is a
+            // no-op, but for a FIFO/socket/chardev write() can return EAGAIN
+            // and we must wait for POLLOUT rather than spin. One fstat on the
+            // just-opened fd is cheap and tells us which we have.
+            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) => {
+    it("delivers every byte to the reader", async () => {
+      using dir = tempDir(`bun-write-fifo-${label}`, {});
+      const fifo = join(String(dir), "f.fifo");
+      mkfifo(fifo);
+
+      // 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";
+
+      // Reader: a node:fs read stream opens O_RDONLY in the thread pool
+      // (blocks there until our write side opens) and drains to EOF.
+      const { promise: opened, resolve: onOpen, reject: onOpenErr } = Promise.withResolvers();
+      const { promise: drained, resolve: onEnd, reject: onReadErr } = Promise.withResolvers();
+      const chunks = [];
+      const reader = fs
+        .createReadStream(fifo)
+        .once("open", onOpen)
+        .on("data", c => chunks.push(c))
+        .once("end", () => onEnd(Buffer.concat(chunks)))
+        .once("error", e => {
+          onOpenErr(e);
+          onReadErr(e);
+        });
+
+      try {
+        // Our O_WRONLY|O_NONBLOCK open returns ENXIO until the reader's open
+        // has started. Retry the write until the reader is attached instead
+        // of sleeping for a guess.
+        let written;
+        while (true) {
+          try {
+            written = await Bun.write(fifo, toSource(payload));
+            break;
+          } catch (e) {
+            if (e?.code !== "ENXIO") throw e;
+            await Bun.sleep(0);
+          }
+        }
+        await opened;
+
+        const got = await drained;
+        expect({ bytes: got.length, tail: got.subarray(-3).toString(), written }).toEqual({
+          bytes: N,
+          tail: "END",
+          written: N,
+        });
+      } finally {
+        reader.destroy();
+      }
+    });
+  });
 });