Bun.write: stream ReadableStream sources instead of stringifying them

docs/runtime/file-io.mdx

@@ -95,6 +95,8 @@ The second argument is the data to write. It can be any of the following:
 - `ArrayBuffer` or `SharedArrayBuffer`
 - `TypedArray` (`Uint8Array`, et. al.)
 - `Response`
+- `ReadableStream`
+- `AsyncIterable` (including async generators)
 
 Bun handles each combination with the fastest available system call on the current platform.
 
@@ -279,7 +281,15 @@ interface Bun {
 
   write(
     destination: string | number | BunFile | URL,
-    input: string | Blob | ArrayBuffer | SharedArrayBuffer | TypedArray | Response,
+    input:
+      | string
+      | Blob
+      | ArrayBuffer
+      | SharedArrayBuffer
+      | TypedArray
+      | Response
+      | ReadableStream
+      | AsyncIterable<string | ArrayBuffer | ArrayBufferView>,
   ): Promise<number>;
 }
 

packages/bun-types/bun.d.ts

@@ -1604,7 +1604,19 @@ declare module "bun" {
    */
   function write(
     destination: BunFile | S3File | PathLike,
-    input: Blob | NodeJS.TypedArray | ArrayBufferLike | string | BlobPart[] | Archive,
+    input:
+      | Blob
+      | NodeJS.TypedArray
+      | ArrayBufferLike
+      | string
+      | BlobPart[]
+      | Archive
+      | ReadableStream
+      | AsyncIterable<string | ArrayBuffer | ArrayBufferView>
+      | AsyncGenerator<string | ArrayBuffer | ArrayBufferView>
+      // must be an `async function*` value; an ordinary function returning
+      // an AsyncGenerator is not converted (same as Response/BodyInit)
+      | (() => AsyncGenerator<string | ArrayBuffer | ArrayBufferView>),
     options?: {
       /**
        * If writing to a PathLike, set the permissions of the file.
@@ -2210,7 +2222,20 @@ declare module "bun" {
      * @param options - The options to use for the write.
      */
     write(
-      data: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer | Request | Response | BunFile,
+      data:
+        | string
+        | ArrayBufferView
+        | ArrayBuffer
+        | SharedArrayBuffer
+        | Request
+        | Response
+        | BunFile
+        | ReadableStream
+        | AsyncIterable<string | ArrayBuffer | ArrayBufferView>
+        | AsyncGenerator<string | ArrayBuffer | ArrayBufferView>
+        // must be an `async function*` value; an ordinary function returning
+        // an AsyncGenerator is not converted (same as Response/BodyInit)
+        | (() => AsyncGenerator<string | ArrayBuffer | ArrayBufferView>),
       options?: { highWaterMark?: number },
     ): Promise<number>;
 

packages/bun-types/s3.d.ts

@@ -724,7 +724,13 @@ declare module "bun" {
         | BunFile
         | S3File
         | Blob
-        | Archive,
+        | Archive
+        | ReadableStream
+        | AsyncIterable<string | ArrayBuffer | ArrayBufferView>
+        | AsyncGenerator<string | ArrayBuffer | ArrayBufferView>
+        // must be an `async function*` value; an ordinary function returning
+        // an AsyncGenerator is not converted (same as Response/BodyInit)
+        | (() => AsyncGenerator<string | ArrayBuffer | ArrayBufferView>),
       options?: S3Options,
     ): Promise<number>;
 
@@ -1058,7 +1064,13 @@ declare module "bun" {
         | S3File
         | Blob
         | File
-        | Archive,
+        | Archive
+        | ReadableStream
+        | AsyncIterable<string | ArrayBuffer | ArrayBufferView>
+        | AsyncGenerator<string | ArrayBuffer | ArrayBufferView>
+        // must be an `async function*` value; an ordinary function returning
+        // an AsyncGenerator is not converted (same as Response/BodyInit)
+        | (() => AsyncGenerator<string | ArrayBuffer | ArrayBufferView>),
       options?: S3Options,
     ): Promise<number>;
 
@@ -1111,7 +1123,13 @@ declare module "bun" {
         | S3File
         | Blob
         | File
-        | Archive,
+        | Archive
+        | ReadableStream
+        | AsyncIterable<string | ArrayBuffer | ArrayBufferView>
+        | AsyncGenerator<string | ArrayBuffer | ArrayBufferView>
+        // must be an `async function*` value; an ordinary function returning
+        // an AsyncGenerator is not converted (same as Response/BodyInit)
+        | (() => AsyncGenerator<string | ArrayBuffer | ArrayBufferView>),
       options?: S3Options,
     ): Promise<number>;
 

src/jsc/bindings/webcore/streams/WebStreamsExports.cpp

@@ -125,6 +125,16 @@ extern "C" bool ReadableStream__isLocked(JSC::EncodedJSValue possibleReadableStr
     return stream && isReadableStreamLocked(stream);
 }
 
+// [[storedError]] when [[state]] is "errored"; empty JSValue otherwise.
+extern "C" JSC::EncodedJSValue ReadableStream__getStoredError(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject*)
+{
+    auto* stream = dynamicDowncast<JSReadableStream>(JSValue::decode(possibleReadableStream));
+    if (!stream || stream->m_state != ReadableStreamState::Errored)
+        return {};
+    JSValue storedError = stream->m_storedError.get();
+    return JSValue::encode(storedError ? storedError : JSC::jsUndefined());
+}
+
 extern "C" void ReadableStream__cancel(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject* globalObject)
 {
     auto* stream = dynamicDowncast<JSReadableStream>(JSValue::decode(possibleReadableStream));

src/runtime/webcore/Blob.rs

@@ -231,6 +231,8 @@ pub trait BlobExt {
         global_this: &JSGlobalObject,
         readable_stream: ReadableStream,
         extra_options: Option<JSValue>,
+        mkdirp_if_not_exists: bool,
+        mode: Option<bun_sys::Mode>,
     ) -> JsResult<JSValue>;
     fn get_writer(&self, global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResult<JSValue>;
     fn get_slice_from(
@@ -1387,6 +1389,8 @@ impl BlobExt for Blob {
         global_this: &JSGlobalObject,
         readable_stream: ReadableStream,
         extra_options: Option<JSValue>,
+        mkdirp_if_not_exists: bool,
+        mode: Option<bun_sys::Mode>,
     ) -> JsResult<JSValue> {
         let Some(store) = self.store.get().clone() else {
             return Ok(
@@ -1467,11 +1471,30 @@ impl BlobExt for Blob {
                 } else {
                     let mut file_path = bun_paths::PathBuffer::uninit();
                     let path = pathlike.path().slice_z(&mut file_path);
-                    match bun_sys::open(
-                        path,
-                        bun_sys::O::WRONLY | bun_sys::O::CREAT | bun_sys::O::NONBLOCK,
-                        WRITE_PERMISSIONS,
-                    ) {
+                    let open_flags = bun_sys::O::WRONLY
+                        | bun_sys::O::CREAT
+                        | bun_sys::O::TRUNC
+                        | bun_sys::O::NONBLOCK;
+                    let open_mode = mode.unwrap_or(WRITE_PERMISSIONS);
+                    let mut opened = bun_sys::open(path, open_flags, open_mode);
+                    if let bun_sys::Result::Err(ref err) = opened {
+                        if err.get_errno() == bun_sys::E::ENOENT && mkdirp_if_not_exists {
+                            match mkdirp_parent_of(path.as_bytes()) {
+                                MkdirpParentResult::Created => {
+                                    opened = bun_sys::open(path, open_flags, open_mode);
+                                }
+                                MkdirpParentResult::Failed(mkdir_err) => {
+                                    // mkdir_err carries the directory path; don't relabel it.
+                                    return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
+                                        global_this,
+                                        mkdir_err.to_js(global_this),
+                                    ));
+                                }
+                                MkdirpParentResult::NoParent => {}
+                            }
+                        }
+                    }
+                    match opened {
                         bun_sys::Result::Ok(result) => result,
                         bun_sys::Result::Err(err) => {
                             return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
@@ -1578,11 +1601,31 @@ impl BlobExt for Blob {
                 let stream_start = streams::Start::FileSink(streams::FileSinkOptions {
                     input_path,
                     chunk_size: 0,
+                    truncate: true,
+                    mode: mode.unwrap_or(WRITE_PERMISSIONS),
                     ..Default::default()
                 });
 
                 // SAFETY: `init` returns a freshly-allocated +1 *mut FileSink.
-                if let bun_sys::Result::Err(err) = unsafe { (*sink).start(&stream_start) } {
+                let mut started = unsafe { (*sink).start(&stream_start) };
+                if let bun_sys::Result::Err(ref err) = started {
+                    if err.get_errno() == bun_sys::E::ENOENT && mkdirp_if_not_exists {
+                        if let PathOrFileDescriptor::Path(p) = &store.data.as_file().pathlike {
+                            match mkdirp_parent_of(p.slice()) {
+                                MkdirpParentResult::Created => {
+                                    // SAFETY: `sink` is still the live +1 from
+                                    // `init`; a failed `start` leaves it reusable.
+                                    started = unsafe { (*sink).start(&stream_start) };
+                                }
+                                MkdirpParentResult::Failed(mkdir_err) => {
+                                    started = bun_sys::Result::Err(mkdir_err);
+                                }
+                                MkdirpParentResult::NoParent => {}
+                            }
+                        }
+                    }
+                }
+                if let bun_sys::Result::Err(err) = started {
                     // SAFETY: release the +1 strong ref taken by `init` on the error path.
                     unsafe { webcore::FileSink::deref(sink) };
                     return Ok(
@@ -1663,18 +1706,21 @@ impl BlobExt for Blob {
                         return Ok(promise_value);
                     }
                     jsc::js_promise::Status::Fulfilled => {
+                        // SAFETY: `file_sink` is still our live +1 ref.
+                        let written = unsafe { (*file_sink).received_bytes.get() };
                         // SAFETY: release our +1 ref on the sink.
                         unsafe { webcore::FileSink::deref(file_sink) };
                         readable_stream.done(global_this);
                         return Ok(JSPromise::resolved_promise_value(
                             global_this,
-                            JSValue::js_number(0.0),
+                            JSValue::js_number(written as f64),
                         ));
                     }
                     jsc::js_promise::Status::Rejected => {
                         // SAFETY: release our +1 ref on the sink.
                         unsafe { webcore::FileSink::deref(file_sink) };
                         readable_stream.cancel(global_this);
+                        promise.set_handled(global_this.vm());
                         return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
                             global_this,
                             promise.result(global_this.vm()),
@@ -1693,12 +1739,17 @@ impl BlobExt for Blob {
                 );
             }
         }
+        // `assignToStream` returned undefined: a direct stream with a
+        // synchronous `pull` already ran to completion inside
+        // `readDirectStream`, so the sink has the byte count.
+        // SAFETY: `file_sink` is still our live +1 ref.
+        let written = unsafe { (*file_sink).received_bytes.get() };
         // SAFETY: release our +1 ref on the sink.
         unsafe { webcore::FileSink::deref(file_sink) };
 
         Ok(JSPromise::resolved_promise_value(
             global_this,
-            JSValue::js_number(0.0),
+            JSValue::js_number(written as f64),
         ))
     }
 
@@ -4422,6 +4473,32 @@ pub fn mkdir_if_not_exists<T: MkdirpTarget>(
     Retry::No
 }
 
+pub enum MkdirpParentResult {
+    Created,
+    Failed(bun_sys::Error),
+    NoParent,
+}
+
+/// `mkdir -p` the parent of `dest_path`; the ENOENT retry for `pipe_readable_stream_to_blob`'s open.
+#[inline(never)]
+fn mkdirp_parent_of(dest_path: &[u8]) -> MkdirpParentResult {
+    let Some(dirname) = bun_core::dirname(dest_path) else {
+        return MkdirpParentResult::NoParent;
+    };
+    let mut node_fs = crate::node::fs::NodeFS::default();
+    match node_fs.mkdir_recursive(&crate::node::fs::args::Mkdir {
+        path: crate::node::PathLike::String(bun_ptr::cow_slice::CowSlice::init_unchecked(
+            dirname, false,
+        )),
+        recursive: true,
+        always_return_none: true,
+        ..Default::default()
+    }) {
+        bun_sys::Result::Ok(_) => MkdirpParentResult::Created,
+        bun_sys::Result::Err(err) => MkdirpParentResult::Failed(err),
+    }
+}
+
 /// `bun_sys::Error` only
 /// exposes `with_path(&[u8])`, so route through the
 /// `PathOrFileDescriptor`'s slice when it's a path and leave the error
@@ -4794,6 +4871,8 @@ pub fn write_file_with_source_destination(
                 ctx,
                 stream,
                 options.extra_options,
+                options.mkdirp_if_not_exists.unwrap_or(true),
+                options.mode,
             );
         } else {
             return Ok(
@@ -5281,6 +5360,38 @@ pub fn write_file_internal(
             break 'brk Blob::init_with_store(archive.store_ref().clone(), global_this);
         }
 
+        if let Some(stream) = ReadableStream::from_js(data, global_this)? {
+            if stream.is_disturbed(global_this) {
+                destination_blob.detach();
+                return Err(global_this.throw_invalid_arguments(format_args!(
+                    "ReadableStream has already been used"
+                )));
+            }
+            // Preflight locked/errored so the pipe's `O_TRUNC` open doesn't run first.
+            if stream.is_locked(global_this) {
+                destination_blob.detach();
+                return Err(
+                    global_this.throw_invalid_arguments(format_args!("ReadableStream is locked"))
+                );
+            }
+            if let Some(stored_error) = stream.stored_error(global_this) {
+                destination_blob.detach();
+                return Ok(
+                    JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
+                        global_this,
+                        stored_error,
+                    ),
+                );
+            }
+            return destination_blob.pipe_readable_stream_to_blob(
+                global_this,
+                stream,
+                options.extra_options,
+                options.mkdirp_if_not_exists.unwrap_or(true),
+                options.mode,
+            );
+        }
+
         break 'brk Blob::get::<false, false>(global_this, data)?;
     };
     // Detach the source blob on scope exit.
@@ -5966,7 +6077,11 @@ pub fn on_file_stream_resolve_request_stream(
     if let Some(stream) = strong.get(global_this) {
         stream.done(global_this);
     }
-    this.promise.resolve(global_this, JSValue::js_number(0.0))?;
+    // SAFETY: `this.sink` is the live +1 ref released by `FileStreamWrapper`'s
+    // `Drop` when `this` goes out of scope below.
+    let written = unsafe { (*this.sink).received_bytes.get() };
+    this.promise
+        .resolve(global_this, JSValue::js_number(written as f64))?;
     Ok(JSValue::UNDEFINED)
 }