fetch: park an unread body stream instead of buffering it without bound

src/http/AsyncHTTP.rs

@@ -280,7 +280,8 @@ impl<'a> AsyncHTTP<'a> {
         &MAX_SIMULTANEOUS_REQUESTS
     }
 
-    pub fn enable_response_body_streaming(&mut self) {
+    /// A store into the shared signal `Store`, not into `self`.
+    pub fn enable_response_body_streaming(&self) {
         self.signals.store(
             crate::signals::Field::ResponseBodyStreaming,
             true,

src/js/internal-for-testing.ts

@@ -671,8 +671,16 @@ export const isMemoryPressureWatcherInstalled: () => boolean = $newCppFunction(
   0,
 );
 
-export const getEventLoopStats: () => { activeTasks: number; concurrentRef: number; numPolls: number } =
-  $newRustFunction("event_loop.rs", "getActiveTasks", 0);
+export const getEventLoopStats: () => {
+  activeTasks: number;
+  tasks: number;
+  immediateTasks: number;
+  concurrentTasksEmpty: boolean;
+  concurrentRef: number;
+  numPolls: number;
+  loopActive: boolean;
+  eventLoopAlive: boolean;
+} = $newRustFunction("event_loop.rs", "getActiveTasks", 0);
 
 export const hostedGitInfo = {
   parseUrl: $newRustFunction("hosted_git_info.rs", "TestingAPIs.jsParseUrl", 1),

src/jsc/Weak.rs

@@ -137,7 +137,8 @@ impl<T> Weak<T> {
         Some(result)
     }
 
-    pub fn clear(&mut self) {
+    /// Clears the C++ slot; `self` keeps the (now empty) handle, so a shared ref suffices.
+    pub fn clear(&self) {
         let Some(r#ref) = self.r#ref else {
             return;
         };

src/jsc/event_loop.rs

@@ -1270,12 +1270,39 @@ pub fn get_active_tasks(global_object: &JSGlobalObject, _frame: &CallFrame) -> J
     // fields and call &-methods on it for the duration of this host fn.
     let vm_ref = global_object.bun_vm();
     let event_loop = vm_ref.event_loop_shared();
-    let result = JSValue::create_empty_object(global_object, 3);
+    let result = JSValue::create_empty_object(global_object, 8);
     result.put(
         global_object,
         b"activeTasks",
         JSValue::js_number(vm_ref.active_tasks as f64),
     );
+    result.put(
+        global_object,
+        b"tasks",
+        JSValue::js_number(event_loop.tasks.readable_length() as f64),
+    );
+    result.put(
+        global_object,
+        b"immediateTasks",
+        JSValue::js_number(
+            (event_loop.immediate_tasks.len() + event_loop.next_immediate_tasks.len()) as f64,
+        ),
+    );
+    result.put(
+        global_object,
+        b"concurrentTasksEmpty",
+        JSValue::from(event_loop.concurrent_tasks.is_empty()),
+    );
+    result.put(
+        global_object,
+        b"loopActive",
+        JSValue::from(vm_ref.platform_loop_opt().is_some_and(|h| h.is_active())),
+    );
+    result.put(
+        global_object,
+        b"eventLoopAlive",
+        JSValue::from(vm_ref.is_event_loop_alive()),
+    );
     result.put(
         global_object,
         b"concurrentRef",

src/runtime/server/RequestContext.rs

@@ -400,14 +400,13 @@ mod shim {
             .as_ref()
             .is_some_and(|s| matches!(s.data, crate::webcore::blob::store::Data::File(_)))
     }
+    /// The response is done with its natively piped body. A body still mid-stream is
+    /// cancelled: it stayed locked to this response, so nothing else can read it.
     #[inline]
     pub(super) fn byte_stream_unpipe(s: NonNull<ByteStream>) {
-        // The lone caller has just `take()`n the pointer out of
-        // `self.byte_stream`; the allocation is kept alive by
-        // `response_body_readable_stream_ref` (BackRef invariant: pointee
-        // outlives this temporary). R-2: `unpipe_without_deref` takes `&self`
-        // (interior-mutable `JsCell<SinkHandle>`), so shared deref is sufficient.
-        bun_ptr::BackRef::from(s).unpipe_without_deref()
+        // The caller has just `take()`n `self.byte_stream` and still holds
+        // `response_body_readable_stream_ref`, which keeps the pointee alive.
+        bun_ptr::BackRef::from(s).detach_finished_sink()
     }
 }
 use crate::server::DevErrorPage;
@@ -1396,6 +1395,14 @@ where
             return;
         }
 
+        // A natively piped body has nobody left to take it. The deref balances the ref
+        // `do_render_with_body` took for the pipe: `end_chunk`, which releases it otherwise,
+        // cannot run once the response is gone.
+        if let Some(stream) = this.byte_stream.take() {
+            shim::byte_stream_unpipe(stream);
+            this.deref();
+        }
+
         // if we can, free the request now.
         if this.is_dead_request() {
             this.finalize_without_deinit();
@@ -3240,20 +3247,21 @@ where
         // SAFETY: caller passes the live `*mut RequestContext` stored as the
         // sink ctx; `_ref` keeps it alive for this call.
         let this = unsafe { &*this };
+        // The stream has already dropped this sink; `_ref` is the pipe's ref.
+        this.byte_stream.set(None);
 
         if this.is_aborted_or_ended() {
             return;
         }
         if let Some(err) = err {
             // No status committed yet: the upstream producer (e.g. a
             // suspended `HTMLRewriter` transform whose async handler
-            // rejected) failed before any bytes were emitted. Detach the
-            // ByteStream state and hand the error to the server's
-            // `error()` hook so it can supply the response.
+            // rejected) failed before any bytes were emitted. Drop the
+            // stream and hand the error to the server's `error()` hook so it
+            // can supply the response.
             if !this.flags.has_written_status() {
                 let global_this = this.server().global_this();
                 let js_err = err.to_js(global_this);
-                this.byte_stream.set(None);
                 this.response_body_readable_stream_ref
                     .with_mut(|s| s.deinit());
                 this.run_error_handler(js_err);

src/runtime/webcore/ByteStream.rs

@@ -86,6 +86,9 @@ impl readable_stream::SourceContext for ByteStream {
     fn deinit_fn(&mut self) {
         Self::finalize(self)
     }
+    fn wrapper_finalized(&mut self) {
+        self.parent_const().producer.get().consumer_collected();
+    }
     fn drain_internal_buffer(&mut self) -> Vec<u8> {
         Self::drain(self)
     }
@@ -178,6 +181,22 @@ impl ByteStream {
         self.sink_paused.set(false);
     }
 
+    /// The sink is gone before the stream ended (its peer went away). The stream stays
+    /// locked to it, so nobody else can read the rest: close the producer too.
+    pub(crate) fn detach_finished_sink(&self) {
+        if self.has_received_last_chunk.get() {
+            self.unpipe_without_deref();
+        } else {
+            self.cancel_from_sink(None);
+        }
+    }
+
+    /// Bytes delivered that no consumer has taken yet.
+    #[inline]
+    pub fn buffered_len(&self) -> usize {
+        self.buffer.get().len() - self.offset.get()
+    }
+
     /// Sink's drain ack: unpause, push buffered bytes, end if last chunk already arrived.
     pub fn resume(&self) {
         if !self.sink_paused.get() {
@@ -265,8 +284,8 @@ impl ByteStream {
         Vec::<u8>::move_from_list(list)
     }
 
-    /// Called by native fast-paths after wiring `self.sink`. Restores
-    /// producer-side backpressure if it was already dropped (BufferAll).
+    /// Called by native fast-paths after wiring `self.sink`: a consumer now
+    /// waits for bytes, so a parked producer resumes.
     pub fn signal_consumer_attached(&self) {
         self.parent_const().producer.get().start();
     }

src/runtime/webcore/ReadableStream.rs

@@ -781,6 +781,10 @@ pub trait SourceContext: Sized {
         false
     }
 
+    /// The JS wrapper was collected while native refs remain. Runs inside a GC
+    /// sweep: no JS. `ByteStream` tells a parked producer nobody can read it now.
+    fn wrapper_finalized(&mut self) {}
+
     /// `setRefUnrefFn` — default no-op.
     fn set_ref_unref(&mut self, _enable: bool) {}
 
@@ -851,6 +855,10 @@ pub struct NewSource<C: SourceContext> {
     /// `Finalized` so [`Self::on_js_close`] reads `None` instead of a
     /// dead-but-unswept cell.
     pub this_jsvalue: jsc::JsRef,
+    /// The producer holding a native ref has parked ([`Self::unroot_wrapper`]):
+    /// its ref keeps this allocation, not the wrapper, so an unread stream can
+    /// be collected. Cleared by [`Self::root_wrapper`].
+    pub wrapper_unrooted: Cell<bool>,
     /// R-2: written by context methods (`ByteStream::to_any_blob`,
     /// `ByteBlobLoader::to_any_blob`) through their parent accessor, so
     /// interior-mutable.
@@ -869,6 +877,7 @@ impl<C: SourceContext + Default> Default for NewSource<C> {
             producer: Cell::new(streams::SourceHandle::None),
             global_this: None,
             this_jsvalue: jsc::JsRef::empty(),
+            wrapper_unrooted: Cell::new(false),
             is_closed: Cell::new(false),
         }
     }
@@ -1140,13 +1149,37 @@ impl<C: SourceContext> NewSource<C> {
         // `waiting_for_on_reader_done` I/O ref). Root the wrapper so
         // `on_js_close`, reached from `on_reader_done` off the event loop with
         // no JS frame on the stack, never reads a dead-but-unswept cell.
+        if !self.wrapper_unrooted.get() {
+            self.upgrade_wrapper();
+        }
+    }
+
+    fn upgrade_wrapper(&mut self) {
         if let Some(global) = self.global_this.as_deref() {
             if self.this_jsvalue.is_not_empty() {
                 self.this_jsvalue.upgrade(global);
             }
         }
     }
 
+    /// The producer keeps its native ref but stops rooting the wrapper: nothing
+    /// is reading, so the stream should be collectable. [`SourceContext::wrapper_finalized`]
+    /// tells the producer if that happens.
+    /// Same access pattern as [`Self::increment_count`]: reached through the
+    /// producer's raw pointer while the context may be borrowed.
+    pub fn unroot_wrapper(&mut self) {
+        self.wrapper_unrooted.set(true);
+        self.this_jsvalue.downgrade();
+    }
+
+    /// Undo [`Self::unroot_wrapper`]: a consumer is reading again.
+    pub fn root_wrapper(&mut self) {
+        self.wrapper_unrooted.set(false);
+        if self.ref_count > 1 {
+            self.upgrade_wrapper();
+        }
+    }
+
     /// Release one reference. If the count hits zero, runs context teardown and
     /// **frees the allocation**.
     ///
@@ -1440,6 +1473,13 @@ impl<C: SourceContext> NewSource<C> {
         let this = Box::into_raw(self);
         // SAFETY: `this` is live — just unwrapped from `Box`.
         unsafe { (*this).this_jsvalue.finalize() };
+        // SAFETY: `this` is live; the JS-wrapper +1 (released last) keeps ref_count > 0
+        // across whatever ref the producer drops in response.
+        unsafe {
+            if (*this).ref_count > 1 {
+                (*this).context.wrapper_finalized();
+            }
+        }
         // SAFETY: `this` is live; the JS-wrapper ref below still pins the count.
         if unsafe { (*this).context.finalize_detach() } {
             // SAFETY: `this` is live; the JS-wrapper +1 (released below) keeps ref_count > 0.