fetch/S3: one high-water-mark rule for response-body backpressure; Bun.write(dest, response) streams to disk

packages/bun-types/bun.d.ts

@@ -2134,7 +2134,7 @@ 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,
     options?: {
       /**
        * If writing to a PathLike, set the permissions of the file.
@@ -2152,19 +2152,20 @@ declare module "bun" {
   ): Promise<number>;
 
   /**
-   * Persist a {@link Response} body to disk.
+   * Persist a {@link Response} or {@link Request} body to disk. The body is
+   * streamed into the file as it arrives.
    *
    * @param destination The file to write to. If the file doesn't exist, it is
    * created; if it does, it is overwritten. If `input` is smaller than
    * `destination`, `destination` is truncated.
-   * @param input The `Response` whose body is written
+   * @param input The `Response` or `Request` whose body is written
    * @param options Options for the write
    *
    * @returns A promise that resolves with the number of bytes written.
    */
   function write(
     destination: BunFile,
-    input: Response,
+    input: Response | Request,
     options?: {
       /**
        * If `true`, create the parent directory if it doesn't exist.
@@ -2178,17 +2179,18 @@ declare module "bun" {
   ): Promise<number>;
 
   /**
-   * Persist a {@link Response} body to disk.
+   * Persist a {@link Response} or {@link Request} body to disk. The body is
+   * streamed into the file as it arrives.
    *
    * @param destinationPath The file path to write to. If the file doesn't
    * exist, it is created; if it does, it is overwritten. If `input` is
    * smaller than the existing file, the file is truncated.
-   * @param input The `Response` whose body is written
+   * @param input The `Response` or `Request` whose body is written
    * @returns A promise that resolves with the number of bytes written.
    */
   function write(
     destinationPath: PathLike,
-    input: Response,
+    input: Response | Request,
     options?: {
       /**
        * If `true`, create the parent directory if it doesn't exist.
@@ -2740,7 +2742,7 @@ 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,
       options?: { highWaterMark?: number },
     ): Promise<number>;
 

src/http/Signals.rs

@@ -12,17 +12,22 @@ pub struct Signals {
     pub body_receive_mode: Option<NonNull<AtomicU8>>,
 }
 
+/// Receive backpressure high-water mark: bytes no consumer has taken, on either side of the
+/// HTTP→JS hop. A body shorter than this completes unread, which frees its connection.
+pub const BODY_HIGH_WATER_MARK: usize = 256 * 1024;
+
+/// Receive backpressure for a body handed to JS. Whichever side holds bytes no consumer has
+/// taken moves `Flowing -> Paused` once they reach the high-water mark; whoever takes them
+/// moves `Paused -> Flowing` and schedules a resume. The transport applies `Paused` after the
+/// next read. Two terminal states: `BufferAll` (a consumer wants the whole body) and
+/// `Abandoned` (nothing will read it; the transport is being shut down, drop what arrives).
 #[repr(u8)]
 #[derive(Copy, Clone, PartialEq, Eq, Debug)]
 pub enum BodyReceiveMode {
-    /// Pause the transport after each delivered body chunk until JS pulls.
-    AutoPause = 0,
-    /// `callback` won the CAS; transport should be paused until JS pulls.
+    Flowing = 0,
     Paused = 1,
-    /// `.arrayBuffer()`/`.text()`/etc attached — never pause.
     BufferAll = 2,
-    /// Cancelled or abandoned — never pause, callback discards bytes.
-    Ignore = 3,
+    Abandoned = 3,
 }
 
 impl BodyReceiveMode {
@@ -31,8 +36,8 @@ impl BodyReceiveMode {
         match v {
             1 => Self::Paused,
             2 => Self::BufferAll,
-            3 => Self::Ignore,
-            _ => Self::AutoPause,
+            3 => Self::Abandoned,
+            _ => Self::Flowing,
         }
     }
 }
@@ -96,7 +101,7 @@ impl Default for Store {
             response_body_streaming: AtomicBool::new(false),
             aborted: AtomicBool::new(false),
             cert_errors: AtomicBool::new(false),
-            body_receive_mode: AtomicU8::new(BodyReceiveMode::AutoPause as u8),
+            body_receive_mode: AtomicU8::new(BodyReceiveMode::Flowing as u8),
         }
     }
 }
@@ -125,21 +130,38 @@ impl Store {
     }
 
     #[inline]
-    pub fn try_transition_receive_mode(&self, from: BodyReceiveMode, to: BodyReceiveMode) -> bool {
+    fn try_transition_receive_mode(&self, from: BodyReceiveMode, to: BodyReceiveMode) -> bool {
         self.body_receive_mode
             .compare_exchange(from as u8, to as u8, Ordering::AcqRel, Ordering::Relaxed)
             .is_ok()
     }
 
-    /// Unconditionally move to a terminal mode (`BufferAll`/`Ignore`).
-    /// Returns whether the previous state was `Paused`.
+    /// `Flowing -> Paused`. No-op in the other states.
+    #[inline]
+    pub fn pause_receive(&self) {
+        let _ = self.try_transition_receive_mode(BodyReceiveMode::Flowing, BodyReceiveMode::Paused);
+    }
+
+    /// `Paused -> Flowing`. Returns whether it was paused, i.e. whether the caller has to
+    /// schedule the transport's resume.
+    #[inline]
+    pub fn unpause_receive(&self) -> bool {
+        self.try_transition_receive_mode(BodyReceiveMode::Paused, BodyReceiveMode::Flowing)
+    }
+
+    /// Terminal: never pause again. Returns whether it was paused.
+    #[inline]
+    pub fn receive_all(&self) -> bool {
+        self.body_receive_mode
+            .swap(BodyReceiveMode::BufferAll as u8, Ordering::AcqRel)
+            == BodyReceiveMode::Paused as u8
+    }
+
+    /// Terminal.
     #[inline]
-    pub fn set_receive_mode_terminal(&self, mode: BodyReceiveMode) -> bool {
-        debug_assert!(matches!(
-            mode,
-            BodyReceiveMode::BufferAll | BodyReceiveMode::Ignore
-        ));
-        self.body_receive_mode.swap(mode as u8, Ordering::AcqRel) == BodyReceiveMode::Paused as u8
+    pub fn abandon(&self) {
+        self.body_receive_mode
+            .store(BodyReceiveMode::Abandoned as u8, Ordering::Release);
     }
 }
 

src/http/h2_client/ClientSession.rs

@@ -683,8 +683,8 @@ impl ClientSession {
         self.by_http_id.get(&async_http_id).copied()
     }
 
-    /// JS just enabled `response_body_streaming` on the request, so flush any
-    /// body bytes that arrived between metadata delivery and `getReader()`.
+    /// A body consumer attached on the JS side: flush any body bytes that arrived between
+    /// metadata delivery and `getReader()`.
     fn drain_response_body(&mut self, async_http_id: u32) {
         let Some(stream) = self.stream_for_http_id(async_http_id) else {
             return;

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

@@ -1017,6 +1017,10 @@ static std::optional<bool> rsisWriteChunk(JSC::VM& vm, JSGlobalObject* globalObj
     bool shouldSuspend = wrote.isNumber() && wrote.asNumber() < 0;
     if (auto* wrotePromise = dynamicDowncast<JSPromise>(wrote)) {
         markPromiseAsHandled(vm, wrotePromise);
+        if (wrotePromise->status() == JSPromise::Status::Rejected) {
+            throwException(globalObject, scope, wrotePromise->result());
+            return std::nullopt;
+        }
         shouldSuspend = wrotePromise->status() == JSPromise::Status::Pending;
     }
     if (shouldSuspend) {