fs: back createWriteStream(path) with a FileSink that adopts the fd

packages/bun-types/bun.d.ts

@@ -1938,6 +1938,11 @@ declare module "bun" {
     }): void;
 
     write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number;
+    /**
+     * Write multiple chunks of data in order.
+     * @returns Number of bytes written
+     */
+    writev(chunks: ArrayBufferView[]): number;
     /**
      * Flush the internal buffer.
      *

packages/bun-types/s3.d.ts

@@ -14,6 +14,15 @@ declare module "bun" {
      * @returns Number of bytes written or, if the write is pending, a Promise resolving to the number of bytes
      */
     write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number | Promise<number>;
+    /**
+     * Write multiple chunks of data to the file in order.
+     *
+     * If the file descriptor is not writable yet, the data is buffered.
+     *
+     * @param chunks The data to write
+     * @returns Number of bytes written or, if the write is pending, a Promise resolving to the number of bytes
+     */
+    writev(chunks: ArrayBufferView[]): number | Promise<number>;
     /**
      * Flush the internal buffer, committing the data to disk or the pipe.
      *

src/codegen/generate-jssink.ts

@@ -1072,7 +1072,7 @@ pub use ${rustPath} as ${name};
 
 `;
 
-    const hostFns = ["construct", "write", "end", "flush", "start"] as const;
+    const hostFns = ["construct", "write", "writev", "end", "flush", "start"] as const;
     for (const fn of hostFns) {
       const sym = `${name}__${fn}`;
       symbols.push(sym);
@@ -1180,6 +1180,7 @@ function lutInput() {
     end        ${`${name}__end`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0
     start      ${`${name}__start`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1
     write      ${`${name}__write`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1
+    writev     ${`${name}__writev`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1
     ref        ${`${name}__ref`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0
     unref      ${`${name}__unref`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0
     _getFd      ${`${name}__getFd`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0
@@ -1194,6 +1195,7 @@ function lutInput() {
     end          ${`${controller}__end`.padEnd(protopad + 4)}  ReadOnly|DontDelete|Function 0
     start        ${`${name}__start`.padEnd(protopad + 4)}  ReadOnly|DontDelete|Function 1
     write        ${`${name}__write`.padEnd(protopad + 4)}  ReadOnly|DontDelete|Function 1
+    writev       ${`${name}__writev`.padEnd(protopad + 4)}  ReadOnly|DontDelete|Function 1
 @end
 */
 `;

src/io/PipeWriter.rs

@@ -67,15 +67,12 @@ pub trait PosixPipeWriter {
     fn try_write(&self, force_sync: bool, buf: &[u8]) -> WriteResult {
         // PERF: try_write_with_write_fn is not monomorphized per FileType —
         // profile if hot.
-        let ft = if !force_sync {
-            self.get_file_type()
-        } else {
-            FileType::File
-        };
-        match ft {
-            FileType::NonblockingPipe | FileType::File => {
-                self.try_write_with_write_fn(buf, sys::write)
-            }
+        if force_sync {
+            return self.try_write_with_write_fn(buf, sys::write);
+        }
+        match self.get_file_type() {
+            FileType::NonblockingPipe => self.try_write_with_write_fn(buf, sys::write),
+            FileType::File => self.try_write_with_write_fn(buf, sys::write_nonblocking),
             FileType::Pipe => self.try_write_with_write_fn(buf, write_to_blocking_pipe),
             FileType::Socket => self.try_write_with_write_fn(buf, sys::send_non_block),
         }
@@ -604,6 +601,7 @@ pub struct PosixStreamingWriter<Parent: PosixStreamingWriterParent> {
     pub is_done: bool,
     pub closed_without_reporting: bool,
     pub force_sync: bool,
+    pub close_fd: bool,
 }
 
 impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent> {
@@ -615,6 +613,7 @@ impl<Parent: PosixStreamingWriterParent> Default for PosixStreamingWriter<Parent
             is_done: false,
             closed_without_reporting: false,
             force_sync: false,
+            close_fd: true,
         }
     }
 }
@@ -766,7 +765,8 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         if self.get_fd() != Fd::INVALID {
             debug_assert!(!self.closed_without_reporting);
             self.closed_without_reporting = true;
-            self.handle.close(None, None::<fn(*mut c_void)>);
+            self.handle
+                .close_impl(None, None::<fn(*mut c_void)>, self.close_fd);
         }
     }
 
@@ -946,6 +946,125 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         rc
     }
 
+    fn writev_buffered(&mut self, bufs: &[&[u8]], total: usize) -> WriteResult {
+        if self.outgoing.ensure_unused_capacity(total).is_err() {
+            return WriteResult::Err(sys::Error::oom());
+        }
+        for b in bufs {
+            self.outgoing.write_assume_capacity(b);
+        }
+        self.maybe_write_newly_buffered_data(total)
+    }
+
+    #[cfg(unix)]
+    fn buffer_tail(&mut self, bufs: &[&[u8]], mut skip: usize, remaining: usize) -> Result<(), ()> {
+        if self.outgoing.ensure_unused_capacity(remaining).is_err() {
+            return Err(());
+        }
+        for b in bufs {
+            if skip >= b.len() {
+                skip -= b.len();
+                continue;
+            }
+            self.outgoing.write_assume_capacity(&b[skip..]);
+            skip = 0;
+        }
+        Ok(())
+    }
+
+    pub fn writev(&mut self, bufs: &[&[u8]]) -> WriteResult {
+        if self.is_done || self.closed_without_reporting {
+            return WriteResult::Done(0);
+        }
+        let mut total: usize = 0;
+        for b in bufs {
+            total += b.len();
+        }
+        if total == 0 {
+            return WriteResult::Wrote(0);
+        }
+        #[cfg(not(unix))]
+        {
+            self.writev_buffered(bufs, total)
+        }
+        #[cfg(unix)]
+        {
+            const IOV_MAX: usize = 1024;
+            if self.outgoing.size() > 0 || self.should_buffer(total) || bufs.len() > IOV_MAX {
+                return self.writev_buffered(bufs, total);
+            }
+
+            let fd = self.get_fd();
+            if fd == Fd::INVALID {
+                return WriteResult::Done(0);
+            }
+            let file_type = if self.force_sync {
+                FileType::File
+            } else {
+                self.get_file_type()
+            };
+            if matches!(file_type, FileType::Pipe) {
+                return self.writev_buffered(bufs, total);
+            }
+
+            let mut iov: Vec<libc::iovec> = Vec::with_capacity(bufs.len());
+            for b in bufs {
+                if b.is_empty() {
+                    continue;
+                }
+                iov.push(libc::iovec {
+                    iov_base: b.as_ptr().cast_mut().cast::<core::ffi::c_void>(),
+                    iov_len: b.len(),
+                });
+            }
+
+            let mut offset: usize = 0;
+            let mut start: usize = 0;
+            loop {
+                let rc = if matches!(file_type, FileType::File) && !self.force_sync {
+                    sys::writev_nonblocking(fd, &iov[start..])
+                } else {
+                    sys::writev(fd, &iov[start..])
+                };
+                let wrote = match rc {
+                    sys::Result::Err(err) => {
+                        if err.is_retry() {
+                            if self.buffer_tail(bufs, offset, total - offset).is_err() {
+                                return WriteResult::Err(sys::Error::oom());
+                            }
+                            self.parent_on_write(offset, WriteStatus::Pending);
+                            Self::register_poll(self);
+                            return WriteResult::Pending(offset);
+                        }
+                        return WriteResult::Err(err);
+                    }
+                    sys::Result::Ok(n) => n,
+                };
+                offset += wrote;
+                if wrote == 0 {
+                    self.parent_on_write(offset, WriteStatus::EndOfFile);
+                    return WriteResult::Done(offset);
+                }
+                if offset == total {
+                    self.parent_on_write(offset, WriteStatus::Drained);
+                    return WriteResult::Wrote(offset);
+                }
+                let mut consumed = wrote;
+                while start < iov.len() && consumed >= iov[start].iov_len {
+                    consumed -= iov[start].iov_len;
+                    start += 1;
+                }
+                if start < iov.len() && consumed > 0 {
+                    // SAFETY: `consumed < iov[start].iov_len`, so the advanced
+                    // base stays within the original live buffer.
+                    iov[start].iov_base =
+                        unsafe { iov[start].iov_base.cast::<u8>().add(consumed) }.cast();
+                    iov[start].iov_len -= consumed;
+                }
+            }
+        }
+    }
+
     pub fn flush(&mut self) -> WriteResult {
         if self.closed_without_reporting || self.is_done {
             return WriteResult::Done(0);
@@ -1026,10 +1145,11 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         }
 
         let parent = self.parent;
-        self.handle.close(
+        self.handle.close_impl(
             Some(parent.cast()),
             // SAFETY: parent was set via set_parent with a *mut Parent.
             Some(|ctx: *mut c_void| unsafe { Parent::on_close(ctx.cast::<Parent>()) }),
+            self.close_fd,
         );
     }
 
@@ -2443,6 +2563,62 @@ impl<Parent: WindowsStreamingWriterParent> WindowsStreamingWriter<Parent> {
         self.write_internal_u8(buffer, WriteKind::Bytes)
     }
 
+    pub fn writev(&mut self, bufs: &[&[u8]]) -> WriteResult {
+        if self.is_done {
+            return WriteResult::Done(0);
+        }
+        let mut total: usize = 0;
+        for b in bufs {
+            total += b.len();
+        }
+        if total == 0 {
+            return WriteResult::Wrote(0);
+        }
+        let had_buffered_data = self.outgoing.is_not_empty();
+        if self.outgoing.ensure_unused_capacity(total).is_err() {
+            return WriteResult::Err(sys::Error::oom());
+        }
+        for b in bufs {
+            self.outgoing.write_assume_capacity(b);
+        }
+
+        if matches!(self.source, Some(Source::SyncFile(_))) {
+            let result = (|| {
+                let remain = self.outgoing.slice();
+                let initial_len = remain.len();
+                let mut remain = remain;
+                let fd = Fd::from_uv(match &self.source {
+                    Some(Source::SyncFile(f)) => f.file,
+                    _ => unreachable!(),
+                });
+                while remain.len() > 0 {
+                    match sys::write(fd, remain) {
+                        sys::Result::Err(err) => return WriteResult::Err(err),
+                        sys::Result::Ok(wrote) => {
+                            remain = &remain[wrote..];
+                            if wrote == 0 {
+                                break;
+                            }
+                        }
+                    }
+                }
+                let wrote = initial_len - remain.len();
+                if wrote == 0 {
+                    return WriteResult::Done(wrote);
+                }
+                WriteResult::Wrote(wrote)
+            })();
+            self.outgoing.reset();
+            return result;
+        }
+
+        if had_buffered_data {
+            return WriteResult::Pending(0);
+        }
+        self.process_send();
+        self.last_write_result.clone()
+    }
+
     pub fn flush(&mut self) -> WriteResult {
         if self.is_done {
             return WriteResult::Done(0);
@@ -2464,7 +2640,8 @@ impl<Parent: WindowsStreamingWriterParent> WindowsStreamingWriter<Parent> {
         self.is_done = true;
 
         if !self.has_pending_data() {
-            if !self.owns_fd {
+            if !self.owns_fd && !matches!(self.source, Some(Source::File(_) | Source::SyncFile(_)))
+            {
                 return;
             }
             self.close();

src/io/openForWriting.rs

@@ -14,6 +14,9 @@ pub trait OpenForWritingInput {
         is_nonblocking: &mut bool,
         openat: &dyn Fn(Fd, &ZStr, i32, Mode) -> bun_sys::Result<Fd>,
     ) -> bun_sys::Result<Fd>;
+    fn borrowed_fd(&self) -> Option<Fd> {
+        None
+    }
 }
 
 impl OpenForWritingInput for crate::PathOrFileDescriptor<'_> {
@@ -31,7 +34,13 @@ impl OpenForWritingInput for crate::PathOrFileDescriptor<'_> {
                 *is_nonblocking = true;
                 bun_sys::openat_a(dir, path, input_flags, mode)
             }
-            Fd(fd_) => bun_sys::dup_with_flags(*fd_, 0),
+            Fd(_) => unreachable!("borrowed_fd() short-circuits Fd in open_for_writing_impl"),
+        }
+    }
+    fn borrowed_fd(&self) -> Option<Fd> {
+        match self {
+            crate::PathOrFileDescriptor::Fd(fd) => Some(*fd),
+            crate::PathOrFileDescriptor::Path(_) => None,
         }
     }
 }
@@ -112,15 +121,30 @@ where
     #[cfg(unix)]
     let mut isatty = false;
     let mut is_nonblocking = false;
-    let result =
-        input_path.open_for_writing_result(dir, input_flags, mode, &mut is_nonblocking, &openat);
-    let fd = result?;
+    let borrowed = input_path.borrowed_fd();
+    let fd = match borrowed {
+        Some(fd) => fd,
+        None => input_path.open_for_writing_result(
+            dir,
+            input_flags,
+            mode,
+            &mut is_nonblocking,
+            &openat,
+        )?,
+    };
+    #[cfg(windows)]
+    let _ = borrowed;
 
     #[cfg(unix)]
     {
+        let close_on_err = |fd: Fd| {
+            if borrowed.is_none() {
+                fd.close();
+            }
+        };
         match bun_sys::fstat(fd) {
             Err(err) => {
-                fd.close();
+                close_on_err(fd);
                 return Err(err);
             }
             Ok(stat) => {
@@ -151,7 +175,7 @@ where
                     let flags = match bun_sys::get_fcntl_flags(fd) {
                         Ok(flags) => flags,
                         Err(err) => {
-                            fd.close();
+                            close_on_err(fd);
                             return Err(err);
                         }
                     };