http: couple fetch() receive backpressure to JS body consumption (h1/h2/h3)

packages/bun-usockets/src/loop.c

@@ -609,10 +609,14 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         // - the socket has hung up, so we will never get more data from it (only applies to macOS, as macOS will send the event the same tick but Linux will not.)
                         // - the event loop isn't very busy, so we can read multiple times in a row
                         #define LOOP_ISNT_VERY_BUSY_THRESHOLD 25
+                        /* Stop if on_data paused us (us_socket_pause from the data
+                         * handler, e.g. fetch() receive backpressure or
+                         * net.Socket#pause) — keep honoring the pause instead of
+                         * pulling bytes the caller asked to defer. */
                         if (
                             s && length >= (LIBUS_RECV_BUFFER_LENGTH - 24 * 1024) && length <= LIBUS_RECV_BUFFER_LENGTH &&
                             (error || loop->num_ready_polls < LOOP_ISNT_VERY_BUSY_THRESHOLD) &&
-                            !us_socket_is_closed(s)
+                            !us_socket_is_closed(s) && !s->flags.is_paused
                         ) {
                             repeat_recv_count += error == 0;
 

src/http/HTTPThread.rs

@@ -122,12 +122,12 @@ pub struct HttpThread {
 
     pub queued_shutdowns: Vec<ShutdownMessage>,
     pub queued_writes: Vec<WriteMessage>,
-    pub queued_response_body_drains: Vec<DrainMessage>,
+    pub queued_receive_resumes: Vec<u32>,
     pub queued_cert_check_resumes: Vec<CertCheckResumeMessage>,
 
     pub queued_shutdowns_lock: Mutex,
     pub queued_writes_lock: Mutex,
-    pub queued_response_body_drains_lock: Mutex,
+    pub queued_receive_resumes_lock: Mutex,
     pub queued_cert_check_resumes_lock: Mutex,
 
     pub queued_threadlocal_proxy_derefs: Vec<*mut ProxyTunnel>,
@@ -177,11 +177,11 @@ impl HttpThread {
             has_pending_queued_abort: false,
             queued_shutdowns: Vec::new(),
             queued_writes: Vec::new(),
-            queued_response_body_drains: Vec::new(),
+            queued_receive_resumes: Vec::new(),
             queued_cert_check_resumes: Vec::new(),
             queued_shutdowns_lock: Mutex::new(),
             queued_writes_lock: Mutex::new(),
-            queued_response_body_drains_lock: Mutex::new(),
+            queued_receive_resumes_lock: Mutex::new(),
             queued_cert_check_resumes_lock: Mutex::new(),
             queued_threadlocal_proxy_derefs: Vec::new(),
             has_awoken: AtomicBool::new(false),
@@ -267,10 +267,6 @@ pub enum WriteMessageType {
     End = 1,
 }
 
-pub struct DrainMessage {
-    pub async_http_id: u32,
-}
-
 pub struct ShutdownMessage {
     pub async_http_id: u32,
 }
@@ -813,51 +809,53 @@ impl HttpThread {
         }
     }
 
-    fn drain_queued_http_response_body_drains(&mut self) {
+    fn drain_queued_receive_resumes(&mut self) {
         loop {
-            // socket.close() can potentially be slow
-            // Let's not block other threads while this runs.
-            let queued_response_body_drains = {
-                let _guard = self.queued_response_body_drains_lock.lock_guard();
-                core::mem::take(&mut self.queued_response_body_drains)
+            let queued = {
+                let _guard = self.queued_receive_resumes_lock.lock_guard();
+                core::mem::take(&mut self.queued_receive_resumes)
             };
-
-            for drain in &queued_response_body_drains {
-                if let Some(socket_ptr) = abort_tracker().get(&drain.async_http_id) {
+            if queued.is_empty() {
+                return;
+            }
+            for id in queued {
+                if let Some(socket_ptr) = abort_tracker().get(&id) {
                     match *socket_ptr {
                         uws::AnySocket::SocketTls(socket) => {
                             let tagged = HTTPContext::<true>::get_tagged_from_socket(socket);
                             if let Some(client) = tagged.client_mut() {
+                                client.resume_receive::<true>(socket);
                                 client.drain_response_body::<true>(socket);
                             }
                             if let Some(session) = tagged.session_mut() {
-                                session.drain_response_body_by_http_id(drain.async_http_id);
+                                let _g = session.ref_scope();
+                                session.resume_receive_by_http_id(id);
+                                session.drain_response_body_by_http_id(id);
                             }
                         }
                         uws::AnySocket::SocketTcp(socket) => {
                             let tagged = HTTPContext::<false>::get_tagged_from_socket(socket);
                             if let Some(client) = tagged.client_mut() {
+                                client.resume_receive::<false>(socket);
                                 client.drain_response_body::<false>(socket);
                             }
                             if let Some(session) = tagged.session_mut() {
-                                session.drain_response_body_by_http_id(drain.async_http_id);
+                                let _g = session.ref_scope();
+                                session.resume_receive_by_http_id(id);
+                                session.drain_response_body_by_http_id(id);
                             }
                         }
                     }
+                } else {
+                    h3::ClientContext::resume_receive_by_http_id(id);
                 }
             }
-            let len = queued_response_body_drains.len();
-            drop(queued_response_body_drains);
-            if len == 0 {
-                break;
-            }
-            bun_core::scoped_log!(HTTPThread, "drained {} queued drains", len);
         }
     }
 
     pub fn drain_events(&mut self) {
         // Process any pending writes **before** aborting.
-        self.drain_queued_http_response_body_drains();
+        self.drain_queued_receive_resumes();
         self.drain_queued_writes();
         self.drain_queued_shutdowns();
         // After shutdowns: an abort or cert-rejection scheduled in the same JS
@@ -959,11 +957,13 @@ impl HttpThread {
         }
     }
 
-    pub fn schedule_response_body_drain(&mut self, async_http_id: u32) {
+    pub fn schedule_receive_resume(&mut self, async_http_id: u32) {
         {
-            let _guard = self.queued_response_body_drains_lock.lock_guard();
-            self.queued_response_body_drains
-                .push(DrainMessage { async_http_id });
+            let _guard = self.queued_receive_resumes_lock.lock_guard();
+            if self.queued_receive_resumes.last() == Some(&async_http_id) {
+                return;
+            }
+            self.queued_receive_resumes.push(async_http_id);
         }
         self.wakeup();
     }

src/http/InternalState.rs

@@ -74,6 +74,7 @@ pub struct InternalStateFlags {
     /// check passed (and implicitly by `InternalState::reset()` on every
     /// redirect hop / failure, so each hop re-parks independently).
     pub is_waiting_for_cert_check: bool,
+    pub receive_paused: bool,
     /// Set once `HTTPClient::compress_body_for_send` has run for this attempt.
     /// Guards header-retry re-entries from compressing again. Cleared by
     /// `reset()`/`init()` so each redirect/retry hop re-compresses from the
@@ -93,6 +94,7 @@ impl InternalStateFlags {
             resend_request_body_on_redirect: false,
             clear_hostname_on_redirect: false,
             is_waiting_for_cert_check: false,
+            receive_paused: false,
             body_compressed: false,
         }
     }

src/http/Signals.rs

@@ -1,5 +1,5 @@
 use core::ptr::NonNull;
-use core::sync::atomic::{AtomicBool, Ordering};
+use core::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 
 #[derive(Default, Clone, Copy)]
 pub struct Signals {
@@ -10,17 +10,35 @@ pub struct Signals {
     pub aborted: Option<NonNull<AtomicBool>>,
     pub cert_errors: Option<NonNull<AtomicBool>>,
     pub upgraded: Option<NonNull<AtomicBool>>,
+    pub body_receive_mode: Option<NonNull<AtomicU8>>,
 }
 
-impl Signals {
-    pub fn is_empty(&self) -> bool {
-        self.aborted.is_none()
-            && self.response_body_streaming.is_none()
-            && self.header_progress.is_none()
-            && self.cert_errors.is_none()
-            && self.upgraded.is_none()
+#[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.
+    Paused = 1,
+    /// `.arrayBuffer()`/`.text()`/etc attached — never pause.
+    BufferAll = 2,
+    /// Cancelled or abandoned — never pause, callback discards bytes.
+    Ignore = 3,
+}
+
+impl BodyReceiveMode {
+    #[inline]
+    pub fn from_u8(v: u8) -> Self {
+        match v {
+            1 => Self::Paused,
+            2 => Self::BufferAll,
+            3 => Self::Ignore,
+            _ => Self::AutoPause,
+        }
     }
+}
 
+impl Signals {
     /// Resolve `field` to a [`BackRef`] over its `AtomicBool` slot, if wired.
     ///
     /// Centralises the back-reference upgrade so [`get`]/[`store`] are
@@ -56,6 +74,13 @@ impl Signals {
             a.store(value, ordering);
         }
     }
+
+    #[inline]
+    pub fn is_receive_paused(self) -> bool {
+        self.body_receive_mode
+            .map(bun_ptr::BackRef::from)
+            .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Paused as u8)
+    }
 }
 
 pub struct Store {
@@ -64,6 +89,7 @@ pub struct Store {
     pub aborted: AtomicBool,
     pub cert_errors: AtomicBool,
     pub upgraded: AtomicBool,
+    pub body_receive_mode: AtomicU8,
 }
 
 impl Default for Store {
@@ -74,6 +100,7 @@ impl Default for Store {
             aborted: AtomicBool::new(false),
             cert_errors: AtomicBool::new(false),
             upgraded: AtomicBool::new(false),
+            body_receive_mode: AtomicU8::new(BodyReceiveMode::AutoPause as u8),
         }
     }
 }
@@ -86,8 +113,39 @@ impl Store {
             aborted: Some(NonNull::from(&self.aborted)),
             cert_errors: Some(NonNull::from(&self.cert_errors)),
             upgraded: Some(NonNull::from(&self.upgraded)),
+            body_receive_mode: None,
         }
     }
+
+    pub fn to_with_backpressure(&mut self) -> Signals {
+        Signals {
+            body_receive_mode: Some(NonNull::from(&self.body_receive_mode)),
+            ..self.to()
+        }
+    }
+
+    #[inline]
+    pub fn body_receive_mode(&self) -> BodyReceiveMode {
+        BodyReceiveMode::from_u8(self.body_receive_mode.load(Ordering::Acquire))
+    }
+
+    #[inline]
+    pub 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`.
+    #[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
+    }
 }
 
 /// Selects one of the atomic flag fields of `Signals`.

src/http/h2_client/ClientSession.rs

@@ -195,7 +195,7 @@ impl ClientSession {
     /// pointer (derived from `&mut self`) carries write provenance for the
     /// final `heap::take` in `deref`.
     #[inline]
-    fn ref_scope(&mut self) -> SessionRefGuard {
+    pub(crate) fn ref_scope(&mut self) -> SessionRefGuard {
         // SAFETY: `self` is a live heap-allocated ClientSession.
         unsafe { SessionRefGuard::new(self) }
     }
@@ -557,6 +557,24 @@ impl ClientSession {
         }
     }
 
+    pub fn resume_receive_by_http_id(&mut self, async_http_id: u32) {
+        let _guard = self.ref_scope();
+        let found = self.streams.values().iter().any(|&s| {
+            stream_mut(s)
+                .client_ref()
+                .is_some_and(|c| c.async_http_id == async_http_id)
+        });
+        if !found {
+            return;
+        }
+        self.replenish_window();
+        if self.write_buffer.is_not_empty() {
+            if let Err(err) = self.flush() {
+                self.fail_all(err);
+            }
+        }
+    }
+
     /// HTTP-thread wake-up from `scheduleRequestWrite`: new body bytes (or
     /// end-of-body) are available in the ThreadSafeStreamBuffer.
     pub fn stream_body_by_http_id(&mut self, async_http_id: u32, ended: bool) {
@@ -612,10 +630,16 @@ impl ClientSession {
         let mut updates: Vec<(u32, u32)> = Vec::new();
         for &s in self.streams.values() {
             let s = stream_mut(s);
-            if s.unacked_bytes >= threshold && !s.remote_closed() {
-                updates.push((s.id, s.unacked_bytes));
-                s.unacked_bytes = 0;
+            if s.unacked_bytes < threshold || s.remote_closed() {
+                continue;
+            }
+            if s.client_ref()
+                .is_some_and(|c| c.signals.is_receive_paused())
+            {
+                continue;
             }
+            updates.push((s.id, s.unacked_bytes));
+            s.unacked_bytes = 0;
         }
         for (id, unacked) in updates {
             self.write_window_update(id, unacked);

src/http/h3_client/ClientContext.rs

@@ -224,4 +224,16 @@ impl ClientContext {
             session_mut(s).stream_body_by_http_id(async_http_id, ended);
         }
     }
+
+    pub fn resume_receive_by_http_id(async_http_id: u32) {
+        let Some(this) = Self::get() else {
+            return;
+        };
+        let ctx = bun_ptr::BackRef::from(this);
+        for &s in ctx.sessions.iter() {
+            if session_mut(s).resume_receive_by_http_id(async_http_id) {
+                return;
+            }
+        }
+    }
 }

src/http/h3_client/ClientSession.rs

@@ -143,6 +143,25 @@ impl ClientSession {
         }
     }
 
+    pub fn resume_receive_by_http_id(&mut self, async_http_id: u32) -> bool {
+        for &stream_ptr in self.pending.iter() {
+            let stream = stream_mut(stream_ptr);
+            let Some(client) = stream.client else {
+                continue;
+            };
+            if client_mut(client).async_http_id != async_http_id {
+                continue;
+            }
+            if core::mem::take(&mut stream.read_paused) {
+                if let Some(qs) = stream.qstream_mut() {
+                    qs.want_read(true);
+                }
+            }
+            return true;
+        }
+        false
+    }
+
     pub(super) fn detach(&mut self, stream: *mut Stream) {
         let st = stream_mut(stream);
         if let Some(cl) = st.client {

src/http/h3_client/Stream.rs

@@ -36,6 +36,7 @@ pub struct Stream {
     pub request_body_done: bool,
     pub is_streaming_body: bool,
     pub headers_delivered: bool,
+    pub read_paused: bool,
 }
 
 impl Stream {
@@ -54,6 +55,7 @@ impl Stream {
             request_body_done: false,
             is_streaming_body: false,
             headers_delivered: false,
+            read_paused: false,
         }))
     }