diff --git a/src/http/AsyncHTTP.rs b/src/http/AsyncHTTP.rs index ca88143e5e34..e46ff76410e9 100644 --- a/src/http/AsyncHTTP.rs +++ b/src/http/AsyncHTTP.rs @@ -30,8 +30,6 @@ bun_core::declare_scope!(AsyncHTTP, visible); pub struct AsyncHTTP<'a> { pub response: Option>, pub request_headers: headers::EntryList, - // Caller-owned response buffer (raw pointer, lifetime-erased); never freed here. - pub response_buffer: *mut MutableString, pub request_body: HTTPRequestBody<'a>, pub(crate) method: Method, pub url: URL<'a>, @@ -302,7 +300,6 @@ impl<'a> AsyncHTTP<'a> { self.elapsed = src.elapsed; self.err = src.err; self.response = src.response; - self.response_buffer = src.response_buffer; self.client.url = src.client.url.clone(); self.client.flags = src.client.flags; self.client.remaining_redirect_count = src.client.remaining_redirect_count; @@ -321,12 +318,9 @@ impl<'a> AsyncHTTP<'a> { // ────────────────────────────────────────────────────────────────────────── struct Preconnect { - // Self-referential — `async_http.response_buffer` borrows - // `self.response_buffer`. `Option` so we can write the field after the heap - // address is fixed (late-init); `None` is never observed after `preconnect()` - // populates it. + // `Option` so we can write the field after the heap address is fixed + // (late-init); `None` is never observed after `preconnect()` populates it. async_http: Option>, - response_buffer: MutableString, url: URL<'static>, is_url_owned: bool, } @@ -336,7 +330,6 @@ impl Preconnect { // SAFETY: `this` was produced by `heap::alloc` in `preconnect()` and is // uniquely owned here; `async_http` was fully written before scheduling. unsafe { - (*this).response_buffer = MutableString::default(); (*this) .async_http .as_mut() @@ -348,7 +341,7 @@ impl Preconnect { free_owned_href((*this).url.href); } // Reclaim and drop the heap allocation (runs Drop on `async_http` - // — which in turn drops `HTTPClient` — and on `response_buffer`). + // — which in turn drops `HTTPClient`). drop(bun_core::heap::take(this)); } } @@ -374,23 +367,20 @@ pub fn preconnect(url: URL<'static>, is_url_owned: bool) { let this: *mut Preconnect = bun_core::heap::into_raw(Box::new(Preconnect { async_http: None, - response_buffer: MutableString::default(), url, is_url_owned, })); // SAFETY: `this` is a freshly Box-allocated, uniquely-owned pointer; we // in-place write `async_http` before any read and before it can be observed - // by another thread. The address of `response_buffer` is stable (heap). + // by another thread. unsafe { - let response_buffer: *mut MutableString = core::ptr::addr_of_mut!((*this).response_buffer); let url = (*this).url.clone(); let async_http = (*this).async_http.insert(AsyncHTTP::init( Method::GET, url, headers::EntryList::default(), b"", - response_buffer, b"", HTTPClientResultCallback::new::(this, Preconnect::on_result), FetchRedirect::Manual, @@ -412,7 +402,6 @@ impl<'a> AsyncHTTP<'a> { url: URL<'a>, headers: headers::EntryList, headers_buf: &'a [u8], - response_buffer: *mut MutableString, request_body: &'a [u8], callback: HTTPClientResultCallback, redirect_type: FetchRedirect, @@ -449,7 +438,6 @@ impl<'a> AsyncHTTP<'a> { let mut this = AsyncHTTP { response: None, request_headers: headers, - response_buffer, request_body: HTTPRequestBody::Bytes(request_body), method, url, @@ -517,7 +505,6 @@ impl<'a> AsyncHTTP<'a> { url: URL<'a>, headers: headers::EntryList, headers_buf: &'a [u8], - response_buffer: *mut MutableString, request_body: &'a [u8], http_proxy: Option>, hostname: Option<&'a [u8]>, @@ -528,7 +515,6 @@ impl<'a> AsyncHTTP<'a> { url, headers, headers_buf, - response_buffer, request_body, noop_callback(), redirect_type, @@ -556,6 +542,7 @@ impl<'a> AsyncHTTP<'a> { pub(crate) struct SingleHTTPChannel { slot: bun_threading::Guarded>>, cv: bun_threading::Condvar, + response_buffer: *mut MutableString, } impl SingleHTTPChannel { @@ -563,6 +550,7 @@ impl SingleHTTPChannel { SingleHTTPChannel { slot: bun_threading::Guarded::new(None), cv: bun_threading::Condvar::new(), + response_buffer: core::ptr::null_mut(), } } fn write_item(&self, item: HTTPClientResult<'static>) { @@ -584,7 +572,7 @@ impl SingleHTTPChannel { fn send_sync_callback( this: *mut SingleHTTPChannel, async_http: *mut AsyncHTTP<'static>, - result: HTTPClientResult<'_>, + mut result: HTTPClientResult<'_>, ) { // `init_sync` leaves every streaming/progress signal unset, so the only // callback is the terminal one; writing on `has_more` would hand @@ -606,25 +594,29 @@ fn send_sync_callback( real.response = None; real.err = async_http.err; real.elapsed = async_http.elapsed; - real.response_buffer = async_http.response_buffer; } - // SAFETY: `this` is the leaked `SingleHTTPChannel` from `send_sync` and is - // alive for the process lifetime; `result` borrows the HTTP-thread copy's - // response buffer, which is the caller's buffer — outlives the read in - // `send_sync`. + // SAFETY: `this` is the heap `SingleHTTPChannel` from `send_sync`; + // `response_buffer` is the caller's `&mut MutableString` which outlives + // `read_item`. unsafe { + result.body_into(&mut (*(*this).response_buffer).list); (*this).write_item(result.detach_lifetime()); } } impl<'a> AsyncHTTP<'a> { - pub fn send_sync(&mut self) -> crate::Result { + pub fn send_sync( + &mut self, + response_buffer: &mut MutableString, + ) -> crate::Result { crate::http_thread::init(&Default::default()); // Note: `Box::leak` is forbidden (PORTING.md §Forbidden); // allocate via `heap::alloc` and reclaim once // the single sync callback has fired and we've read the result. - let ctx = bun_core::heap::into_raw_nn(Box::new(SingleHTTPChannel::init())); + let mut ch = SingleHTTPChannel::init(); + ch.response_buffer = &raw mut *response_buffer; + let ctx = bun_core::heap::into_raw_nn(Box::new(ch)); self.result_callback = HTTPClientResultCallback::new::(ctx.as_ptr(), send_sync_callback); @@ -851,20 +843,12 @@ impl<'a> AsyncHTTP<'a> { self.elapsed = http_thread_timer_read(); - // `response_buffer` was set in `init()` to a caller-owned MutableString - // that outlives this request — the very buffer `start()` records as - // `state.body_out_str`. Route through the shared `body_out` accessor - // (one centralised unsafe). - let response_buffer = crate::body_out::as_mut( - NonNull::new(self.response_buffer).expect("response_buffer set in init"), - ); - // Note: `HTTPRequestBody` is not `Clone` (the `Stream` arm holds an // intrusive refcount). Move owned // payloads into the client and leave a detached placeholder so Drop on // `self.request_body` is a no-op. let body = core::mem::replace(&mut self.request_body, HTTPRequestBody::Bytes(b"")); - self.client.start(body, response_buffer); + self.client.start(body); } } diff --git a/src/http/InternalState.rs b/src/http/InternalState.rs index 693bd8b89fb9..18fbca1d08fe 100644 --- a/src/http/InternalState.rs +++ b/src/http/InternalState.rs @@ -1,5 +1,3 @@ -use core::ptr::NonNull; - use crate::Error; use bun_core::MutableString; use bun_core::Output; @@ -16,7 +14,7 @@ pub struct InternalState<'a> { /// This is the cloned metadata containing the response headers, url and status code after the .headers phase are received /// will be turned None once returned to the user (the ownership is transferred to the user) /// this can happen after await fetch(...) and the body can continue streaming when this is already None - /// the user will receive only chunks of the body stored in body_out_str + /// the user will receive only chunks of the body stored in decoded_body pub(crate) cloned_metadata: Option, pub(crate) flags: InternalStateFlags, @@ -26,9 +24,10 @@ pub struct InternalState<'a> { pub(crate) chunked_decoder: bun_picohttp::phr_chunked_decoder, pub(crate) decompressor: Decompressor, pub(crate) stage: Stage, - /// This is owned by the user and should not be freed here. - /// Non-owning back-reference, kept as a raw `NonNull` (BACKREF per PORTING.md). - pub(crate) body_out_str: Option>, + /// Decoded (post-decompression / post-chunked-decode) body bytes accumulate + /// here. Delivered to the progress callback as a borrowed slice and cleared + /// (cap-bounded) after the callback returns. + pub(crate) decoded_body: MutableString, pub(crate) compressed_body: MutableString, pub(crate) content_length: Option, pub(crate) total_body_received: usize, @@ -124,7 +123,7 @@ impl Default for InternalState<'_> { chunked_decoder: bun_picohttp::phr_chunked_decoder::default(), decompressor: Decompressor::None, stage: Stage::Pending, - body_out_str: None, + decoded_body: MutableString::init_empty(), compressed_body: MutableString::init_empty(), content_length: None, total_body_received: 0, @@ -142,17 +141,14 @@ impl Default for InternalState<'_> { } impl<'a> InternalState<'a> { - pub(crate) fn init( - body: HTTPRequestBody<'a>, - body_out_str: &mut MutableString, - ) -> InternalState<'a> { + pub(crate) fn init(body: HTTPRequestBody<'a>) -> InternalState<'a> { let request_body = bun_ptr::RawSlice::new(body.slice()); InternalState { original_request_body: body, request_body, compressed_body: MutableString::init_empty(), response_message_buffer: MutableString::init_empty(), - body_out_str: Some(NonNull::from(body_out_str)), + decoded_body: MutableString::init_empty(), stage: Stage::Pending, ..Default::default() } @@ -163,16 +159,16 @@ impl<'a> InternalState<'a> { } pub(crate) fn reset(&mut self) { - let body_msg = self.body_out_str; - if let Some(body) = body_msg { - crate::body_out::as_mut(body).reset(); - } + // Preserve `decoded_body` across the reset so the progress-update path + // can deliver its bytes after calling `reset()`; clearing happens in + // the caller after `callback.run()`. + let decoded_body = core::mem::take(&mut self.decoded_body); // `*self = ...` below drops every field via drop glue. Only // `original_request_body` needs an explicit `deinit()` because // `HTTPRequestBody` deliberately has no `Drop` (see HTTPRequestBody.rs). self.original_request_body.deinit(); *self = InternalState { - body_out_str: body_msg, + decoded_body, compressed_body: MutableString::init_empty(), response_message_buffer: MutableString::init_empty(), original_request_body: HTTPRequestBody::Bytes(b""), @@ -186,41 +182,27 @@ impl<'a> InternalState<'a> { /// The buffer response body bytes accumulate into. For compressed /// responses this is the intermediate `compressed_body`; otherwise it is - /// the caller-owned `body_out_str`. When `body_out_str` is `None` (the - /// request is in a transitional/terminal state where no owner buffer is - /// attached) fall back to `compressed_body` so the chunked decoder can - /// still run without panicking; those bytes are discarded on the next - /// `reset()`. + /// the owned `decoded_body`. pub(crate) fn get_body_buffer(&mut self) -> &mut MutableString { if self.encoding.is_compressed() { return &mut self.compressed_body; } - match self.body_out_str { - Some(p) => crate::body_out::as_mut(p), - None => &mut self.compressed_body, - } + &mut self.decoded_body } /// Split-borrow `chunked_decoder` and the body buffer (which is either - /// `compressed_body` or the caller-owned `body_out_str`). Both targets are - /// disjoint from each other and from every other field touched by - /// `phr_decode_chunked` callers, so this lets the chunked-decode hot path - /// in `lib.rs` operate on safe references instead of repeated raw-ptr - /// place expressions. + /// `compressed_body` or `decoded_body`). Both targets are disjoint from + /// each other and from every other field touched by `phr_decode_chunked` + /// callers, so this lets the chunked-decode hot path in `lib.rs` operate + /// on safe references instead of repeated raw-ptr place expressions. #[inline] pub(crate) fn chunked_decoder_and_body_buffer( &mut self, ) -> (&mut bun_picohttp::phr_chunked_decoder, &mut MutableString) { - match self.body_out_str { - _ if self.encoding.is_compressed() => { - (&mut self.chunked_decoder, &mut self.compressed_body) - } - // body_out_str is a separate heap allocation, never aliasing - // `chunked_decoder` (a value field of `self`). - Some(p) => (&mut self.chunked_decoder, crate::body_out::as_mut(p)), - // See `get_body_buffer`: fall back to `compressed_body` rather - // than panic when no owner buffer is attached. - None => (&mut self.chunked_decoder, &mut self.compressed_body), + if self.encoding.is_compressed() { + (&mut self.chunked_decoder, &mut self.compressed_body) + } else { + (&mut self.chunked_decoder, &mut self.decoded_body) } } @@ -263,7 +245,6 @@ impl<'a> InternalState<'a> { pub(crate) fn decompress_bytes( &mut self, buffer: &[u8], - body_out_str: &mut MutableString, is_final_chunk: bool, ) -> Result<(), Error> { // A response that declared a Content-Encoding but sent zero body bytes @@ -315,13 +296,13 @@ impl<'a> InternalState<'a> { if (estimated_size as usize) > deflater.shared_buffer.len() && estimated_size < 32 * 1024 * 1024 { - body_out_str.list.reserve_exact( - (estimated_size as usize).saturating_sub(body_out_str.list.len()), + self.decoded_body.list.reserve_exact( + (estimated_size as usize).saturating_sub(self.decoded_body.list.len()), ); - body_out_str.list.clear(); + self.decoded_body.list.clear(); let result = deflater.decompressor_mut().decompress_to_vec( buffer, - &mut body_out_str.list, + &mut self.decoded_body.list, bun_libdeflate::Encoding::Gzip, ); // libdeflate decodes a single gzip member; unconsumed @@ -332,7 +313,7 @@ impl<'a> InternalState<'a> { { still_needs_to_decompress = false; } else { - body_out_str.list.clear(); + self.decoded_body.list.clear(); } break 'libdeflate; @@ -356,10 +337,10 @@ impl<'a> InternalState<'a> { // libdeflate decodes a single member; unconsumed input means // a multi-member gzip stream. Let the zlib path handle it. if result.status == bun_libdeflate::Status::Success && result.read == buffer.len() { - body_out_str + self.decoded_body .list - .reserve_exact(result.written.saturating_sub(body_out_str.list.len())); - body_out_str + .reserve_exact(result.written.saturating_sub(self.decoded_body.list.len())); + self.decoded_body .list .extend_from_slice(&deflater.shared_buffer[0..result.written]); still_needs_to_decompress = false; @@ -371,21 +352,23 @@ impl<'a> InternalState<'a> { // Slow path, or brotli: use the .decompressor if still_needs_to_decompress { log!("Decompressing {} bytes\n", buffer.len()); - if body_out_str.list.capacity() == 0 { + if self.decoded_body.list.capacity() == 0 { let min = ((buffer.len() as f64) * 1.5) .ceil() .min(1024.0 * 1024.0 * 2.0); - if let Err(err) = body_out_str.grow_by((min as usize).max(32)) { + if let Err(err) = self.decoded_body.grow_by((min as usize).max(32)) { self.compressed_body.reset(); return Err(err.into()); } } let is_done = self.is_done(); - if let Err(err) = - self.decompressor - .decompress_chunk(self.encoding, buffer, body_out_str, is_done) - { + if let Err(err) = self.decompressor.decompress_chunk( + self.encoding, + buffer, + &mut self.decoded_body, + is_done, + ) { if is_done || err != crate::Error::ShortRead { bun_core::pretty_errorln!( "Decompression error: {}", @@ -404,7 +387,7 @@ impl<'a> InternalState<'a> { // `buffer` is always the current body buffer's bytes. To avoid aliased &mut/& under // Stacked Borrows (decompress_bytes mutates `self.compressed_body`; the uncompressed - // path materialises `&mut *body_out_str`), callers `mem::take` the body buffer's `list` + // path materialises `&mut self.decoded_body`), callers `mem::take` the body buffer's `list` // and pass it here as an owned Vec — no `&` into `self` survives across `&mut self`. pub(crate) fn process_body_buffer( &mut self, @@ -418,36 +401,22 @@ impl<'a> InternalState<'a> { return Ok(false); } - // `decompress_bytes` below takes `&mut self` alongside `body_out_str`, - // so a `&mut self` accessor would tie the borrow to `self`. The free - // `body_out::as_mut` yields an unbounded `&mut` to the disjoint - // caller-owned allocation. - let Some(body_out_ptr) = self.body_out_str else { - // No owner buffer attached (see `get_body_buffer`). There is - // nowhere to deliver decoded bytes; put the buffer back so the - // caller's take is a no-op and report no progress. The request - // is already in a transitional/terminal state. - self.get_body_buffer().list = buffer; - return Ok(false); - }; - let body_out_str = crate::body_out::as_mut(body_out_ptr); - match self.encoding { Encoding::Brotli | Encoding::Gzip | Encoding::Deflate | Encoding::Zstd => { - self.decompress_bytes(&buffer, body_out_str, is_final_chunk)?; + self.decompress_bytes(&buffer, is_final_chunk)?; // Retain capacity by // returning the (cleared) allocation to compressed_body instead of dropping it. buffer.clear(); self.compressed_body.list = buffer; } _ => { - // Uncompressed: caller took `buffer` from `body_out_str.list`, leaving it - // empty — move the bytes back. If body_out_str is + // Uncompressed: caller took `buffer` from `decoded_body.list`, leaving it + // empty — move the bytes back. If decoded_body is // somehow non-empty, fall back to append. - if body_out_str.list.is_empty() { - body_out_str.list = buffer; - } else if !body_out_str.owns(&buffer) { - if let Err(err) = body_out_str.append(&buffer) { + if self.decoded_body.list.is_empty() { + self.decoded_body.list = buffer; + } else if !self.decoded_body.owns(&buffer) { + if let Err(err) = self.decoded_body.append(&buffer) { let err: Error = err.into(); bun_core::pretty_errorln!( "Failed to append to body buffer: {}", @@ -460,7 +429,7 @@ impl<'a> InternalState<'a> { } } - Ok(!body_out_str.list.is_empty()) + Ok(!self.decoded_body.list.is_empty()) } } diff --git a/src/http/lib.rs b/src/http/lib.rs index c84bbac58887..7dbf1a65510a 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -329,6 +329,13 @@ pub(crate) const MAX_H2_RETRIES: u8 = 5; const PREALLOCATE_MAX: usize = 1024 * 1024 * 256; +/// Per-chunk scratch buffers (`InternalState::decoded_body` on the HTTP thread +/// and `FetchTasklet::scheduled_response_buffer` on the JS thread) whose +/// capacity has grown past this are dropped after the chunk is consumed rather +/// than `clear()`ed-and-reused, so the per-connection high-water mark stays +/// bounded for long-lived streaming responses. +pub const DECODED_BODY_RETAIN_CAP: usize = 512 * 1024; + /// Whether the experimental Alt-Svc-driven HTTP/3 upgrade is enabled at all /// (CLI flag or env var). Used on its own to gate `H3.AltSvc.record` — a /// response that arrived over a request shape h3 can't serve (proxy, sendfile, @@ -427,7 +434,15 @@ pub enum BodySize { #[derive(Default)] pub struct HTTPClientResult<'a> { - pub body: Option<&'a mut MutableString>, + pub body: &'a [u8], + /// Populated only on the terminal (`!has_more`) progress callback: + /// `send_progress_update_*` moves the whole `decoded_body.list` here so + /// one-shot consumers (`send_sync`, `NetworkTask` manifest, S3 simple, + /// `RemoteImageDownload`) can `mem::take` it instead of + /// `extend_from_slice`ing the borrowed `body`. Streaming consumers read + /// `body` on non-terminal callbacks and treat this as just the final + /// chunk's bytes. + pub body_owned: Vec, pub has_more: bool, pub redirected: bool, pub can_stream: bool, @@ -483,23 +498,45 @@ impl<'a> HTTPClientResult<'a> { ) } - /// Widen the borrow on `body` to `'static` for self-referential storage. + /// Returns this callback's body bytes as a slice regardless of which + /// field carries them (`body` on non-terminal, `body_owned` on terminal). + #[inline] + pub fn body_bytes(&self) -> &[u8] { + if self.body.is_empty() { + self.body_owned.as_slice() + } else { + self.body + } + } + + /// Moves this callback's body bytes into `dest`. On a terminal callback + /// with `dest` empty this is a `Vec` move; otherwise it appends. + #[inline] + pub fn body_into(&mut self, dest: &mut Vec) { + if !self.body.is_empty() { + dest.extend_from_slice(self.body); + } else if !self.body_owned.is_empty() { + if dest.is_empty() { + core::mem::swap(dest, &mut self.body_owned); + } else { + dest.extend_from_slice(&self.body_owned); + } + } + } + + /// Widen the borrow to `'static` for self-referential storage. /// - /// Field-by-field move (no bitwise reinterpret): the only lifetime-carrying - /// field is `body: Option<&'a mut MutableString>`, which always points at a - /// buffer owned by the same heap object that will store this result - /// (`FetchTasklet.response_buffer`, `NetworkTask.response_buffer`, …). + /// `body` is the only lifetime-carrying field; it borrows the HTTP + /// thread's `decoded_body` scratch buffer, which is cleared immediately + /// after the callback returns, so the stored form carries `body: &[]`. /// /// # Safety - /// Caller must guarantee `body`'s pointee outlives the returned value and - /// is not aliased exclusively elsewhere for that duration. + /// Caller must not read `.body` from the returned value. #[inline] pub unsafe fn detach_lifetime(self) -> HTTPClientResult<'static> { HTTPClientResult { - // SAFETY: caller contract — the buffer outlives the stored result. - body: self - .body - .map(|b| unsafe { &mut *core::ptr::from_mut::(b) }), + body: &[], + body_owned: self.body_owned, has_more: self.has_more, redirected: self.redirected, can_stream: self.can_stream, @@ -1557,10 +1594,6 @@ impl<'a> HTTPClient<'a> { self.state.request_body.slice() } #[inline] - fn body_out_str(&self) -> Option<&MutableString> { - body_out::opt_mut(self.state.body_out_str).map(|b| &*b) - } - #[inline] fn proxy_tunnel_mut(&mut self) -> Option<&mut ProxyTunnel> { let raw = self.proxy_tunnel.as_ref().map(|p| p.as_ptr())?; Some(proxy_tunnel::raw_as_mut(raw)) @@ -1587,8 +1620,7 @@ impl<'a> HTTPClient<'a> { /// build the result, reset request state, and dispatch the callback. fn dispatch_result_and_reset(&mut self, clear_proxy_tunneling: bool) { let callback = self.result_callback; - let body = self.state.body_out_str; - let mut result = self.to_result(); + let result = self.to_result(); self.state.reset(); // `state.reset()` returns every stage field to Pending, which makes // this finished client indistinguishable from a fresh one. Every @@ -1606,9 +1638,6 @@ impl<'a> HTTPClient<'a> { if clear_proxy_tunneling { self.flags.proxy_tunneling = false; } - // `state.reset()` cleared the caller-owned body buffer; attach it now - // so the callback sees the (empty) post-reset buffer. - result.body = body_out::opt_mut(body); callback.run(self.parent_async_http(), result); } #[inline] @@ -1633,48 +1662,6 @@ impl<'a> HTTPClient<'a> { } } -/// Module-private accessors for the caller-owned `body_out_str` buffer. -/// -/// `state.body_out_str` is a `NonNull` set in `start()` to a -/// buffer owned by the request initiator (FetchTasklet/NetworkTask/…) that -/// strictly outlives the HTTPClient. The buffer is a separate heap allocation -/// from `HTTPClient`/`InternalState`, so a `&mut MutableString` derived here -/// never overlaps a `&mut self` on the client. -/// -/// Centralising the SAFETY argument removes a dozen open-coded -/// `unsafe { p.as_mut() }` derefs at call sites. -mod body_out { - use super::{MutableString, NonNull}; - - /// Upgrade the body-out NonNull to `&mut MutableString`. - /// INVARIANT (module): `p` was obtained from `state.body_out_str` (or its - /// upstream source, `AsyncHTTP.response_buffer`, which `start()` forwards - /// into `body_out_str`). - #[inline] - pub(crate) fn as_mut<'a>(mut p: NonNull) -> &'a mut MutableString { - // SAFETY: see module-level invariant. - unsafe { p.as_mut() } - } - /// `Option`-lifted [`as_mut`]. - #[inline] - pub(super) fn opt_mut<'a>(p: Option>) -> Option<&'a mut MutableString> { - p.map(as_mut) - } - /// Snapshot the body buffer's contents by value so a following - /// `state.reset()` doesn't deliver an empty body. - #[inline] - pub(super) fn take_list(p: Option>) -> Option> { - p.map(|p| core::mem::take(&mut as_mut(p).list)) - } - /// Restore the body bytes that `state.reset()` cleared. - #[inline] - pub(super) fn restore_list(p: Option>, v: Option>) { - if let (Some(p), Some(v)) = (p, v) { - as_mut(p).list = v; - } - } -} - // ───────────────────────────── impl HTTPClient ───────────────────────────── impl<'a> HTTPClient<'a> { @@ -2025,12 +2012,6 @@ impl<'a> HTTPClient<'a> { pub(crate) fn retry_from_h2(&mut self) { debug_assert!(self.h2.is_none()); self.unregister_abort_tracker(); - // No owner buffer means the request is already terminal (see - // `InternalState::get_body_buffer`); there is nowhere to deliver a - // retried response. - let Some(body_out) = self.state.body_out_str else { - return; - }; self.flags.protocol = Protocol::Http1_1; self.h2_retries += 1; let body = core::mem::replace( @@ -2038,7 +2019,7 @@ impl<'a> HTTPClient<'a> { HTTPRequestBody::Bytes(b""), ); self.state.reset(); - self.start(body, body_out::as_mut(body_out)); + self.start(body); } /// Called by the HTTP/2 session for stream-level termination (RST_STREAM, @@ -2104,19 +2085,14 @@ impl<'a> HTTPClient<'a> { && self.state.response_stage != ResponseStage::Body && self.state.response_stage != ResponseStage::BodyChunk { - // No owner buffer means the request is already terminal (see - // `InternalState::get_body_buffer`); there is nowhere to deliver - // a retried response. - if let Some(body_out) = self.state.body_out_str { - self.allow_retry = false; - // we need to retry the request, clean up the response message buffer and start again - self.state.response_message_buffer = MutableString::default(); - let body = core::mem::replace( - &mut self.state.original_request_body, - HTTPRequestBody::Bytes(b""), - ); - self.start(body, body_out::as_mut(body_out)); - } + self.allow_retry = false; + // we need to retry the request, clean up the response message buffer and start again + self.state.response_message_buffer = MutableString::default(); + let body = core::mem::replace( + &mut self.state.original_request_body, + HTTPRequestBody::Bytes(b""), + ); + self.start(body); return; } @@ -2597,14 +2573,6 @@ impl<'a> HTTPClient<'a> { self.state.response_message_buffer = MutableString::default(); - // Copy the NonNull, do NOT `.take()` — the TooManyRedirects `fail()` - // below still needs a populated body pointer. No owner buffer means - // the request is already terminal; there is nowhere to deliver a - // redirected response. - let Some(body_out_str) = self.state.body_out_str else { - GenHttpContext::::close_socket(socket); - return; - }; self.remaining_redirect_count = self.remaining_redirect_count.saturating_sub(1); self.flags.redirected = true; debug_assert!(self.redirect_type == FetchRedirect::Follow); @@ -2676,10 +2644,7 @@ impl<'a> HTTPClient<'a> { self.flags.protocol = Protocol::Http1_1; self.reevaluate_proxy_for_redirect(); - self.start( - HTTPRequestBody::Bytes(request_body), - body_out::as_mut(body_out_str), - ); + self.start(HTTPRequestBody::Bytes(request_body)); } /// Re-resolve `http_proxy` against the post-redirect `self.url`. The @@ -2718,11 +2683,9 @@ impl<'a> HTTPClient<'a> { self.url.is_https() } - pub(crate) fn start(&mut self, body: HTTPRequestBody<'a>, body_out_str: &mut MutableString) { - body_out_str.reset(); - + pub(crate) fn start(&mut self, body: HTTPRequestBody<'a>) { debug_assert!(self.state.response_message_buffer.list.capacity() == 0); - self.state = InternalState::init(body, body_out_str); + self.state = InternalState::init(body); if self.is_https() { self.start_::(); @@ -4113,10 +4076,7 @@ impl<'a> HTTPClient<'a> { return; } - let Some(body_out_str) = self.body_out_str() else { - return; - }; - if body_out_str.list.is_empty() { + if self.state.decoded_body.list.is_empty() { // No update! Don't do anything. return; } @@ -4133,11 +4093,6 @@ impl<'a> HTTPClient<'a> { if self.flags.protocol != Protocol::Http1_1 { return self.send_progress_update_multiplexed(); } - let body = self.state.body_out_str; - // Snapshot the body buffer's CONTENTS by value so that `state.reset()` - // — which calls `body.reset()` and clears the list — doesn't deliver - // an empty body when `is_done`. Restored below before the callback. - let body_snapshot = body_out::take_list(body); let callback = self.result_callback; let mut result = self.to_result(); @@ -4266,13 +4221,23 @@ impl<'a> HTTPClient<'a> { bun_core::scoped_log!(fetch, "done"); } - // Restore the body bytes that `state.reset()` cleared. - body_out::restore_list(body, body_snapshot); - result.body = body_out::opt_mut(body); - callback.run(self.parent_async_http(), result); - + // Move the body bytes out of `self.state` before `callback.run`, since + // `on_async_http_callback_raw` on the terminal callback drops + // `self.state` and deallocates the embedding `ThreadlocalAsyncHTTP` + // (making `self` dangle) before dispatching to the user callback. + let mut decoded_body = core::mem::take(&mut self.state.decoded_body); + let parent = self.parent_async_http(); if has_more { + result.body = decoded_body.list.as_slice(); + callback.run(parent, result); + if decoded_body.list.capacity() <= DECODED_BODY_RETAIN_CAP { + decoded_body.list.clear(); + self.state.decoded_body = decoded_body; + } self.maybe_pause_receive(socket); + } else { + result.body_owned = decoded_body.list; + callback.run(parent, result); } if PRINT_EVERY != 0 { @@ -4291,9 +4256,6 @@ impl<'a> HTTPClient<'a> { /// transport, so there is no `ctx`/`socket` to hand back to the pool here. fn send_progress_update_multiplexed(&mut self) { debug_assert!(self.flags.protocol != Protocol::Http1_1); - let body = self.state.body_out_str; - // Snapshot the body buffer's CONTENTS by value; restored below. - let body_snapshot = body_out::take_list(body); let callback = self.result_callback; let mut result = self.to_result(); @@ -4307,10 +4269,21 @@ impl<'a> HTTPClient<'a> { self.state.stage = Stage::Done; self.flags.proxy_tunneling = false; } - // Restore the body bytes that `state.reset()` cleared. - body_out::restore_list(body, body_snapshot); - result.body = body_out::opt_mut(body); - callback.run(self.parent_async_http(), result); + // See `send_progress_update_without_stage_check`: move the body out of + // `self.state` before the terminal callback's `self` teardown. + let mut decoded_body = core::mem::take(&mut self.state.decoded_body); + let parent = self.parent_async_http(); + if is_done { + result.body_owned = decoded_body.list; + callback.run(parent, result); + return; + } + result.body = decoded_body.list.as_slice(); + callback.run(parent, result); + if decoded_body.list.capacity() <= DECODED_BODY_RETAIN_CAP { + decoded_body.list.clear(); + self.state.decoded_body = decoded_body; + } } /// `do_redirect` minus the per-request socket release/close. The session @@ -4346,13 +4319,6 @@ impl<'a> HTTPClient<'a> { b"" }; self.state.response_message_buffer = MutableString::default(); - // Copy the NonNull, do NOT `.take()` — the TooManyRedirects `fail()` - // below still needs a populated body pointer. No owner buffer means - // the request is already terminal; there is nowhere to deliver a - // redirected response. - let Some(body_out_str) = self.state.body_out_str else { - return; - }; self.remaining_redirect_count = self.remaining_redirect_count.saturating_sub(1); self.flags.redirected = true; debug_assert!(self.redirect_type == FetchRedirect::Follow); @@ -4367,11 +4333,7 @@ impl<'a> HTTPClient<'a> { self.flags.proxy_tunneling = false; self.flags.protocol = Protocol::Http1_1; self.reevaluate_proxy_for_redirect(); - // SAFETY: body_out_str points at the caller-owned MutableString. - self.start( - HTTPRequestBody::Bytes(request_body), - body_out::as_mut(body_out_str), - ); + self.start(HTTPRequestBody::Bytes(request_body)); } pub(crate) fn progress_update_h3(&mut self) { @@ -4457,11 +4419,10 @@ impl<'a> HTTPClient<'a> { /// Build the result payload for the progress/completion callback. /// - /// `body` is left `None`: every caller attaches it from `state.body_out_str` - /// *after* the `state.reset()` that follows this call (reset writes through - /// the same allocation). With `body` absent the result is fully owned, so - /// it can be held across the caller's `&mut self` mutations without a - /// lifetime widen. + /// `body` is left `&[]`: every caller attaches it from + /// `state.decoded_body` *after* the `state.reset()` that follows this + /// call. With `body` empty the result has no borrow into `self`, so it + /// can be held across the caller's `&mut self` mutations. pub(crate) fn to_result(&mut self) -> HTTPClientResult<'static> { let body_size: BodySize = if self.state.is_chunked_encoding() { BodySize::TotalReceived(self.state.total_body_received) @@ -4483,7 +4444,8 @@ impl<'a> HTTPClient<'a> { // transfer ownership of the metadata here return HTTPClientResult { metadata: Some(metadata), - body: None, + body: &[], + body_owned: Vec::new(), redirected: self.flags.redirected, fail: self.state.fail, dns_error: self.state.dns_error, @@ -4499,7 +4461,8 @@ impl<'a> HTTPClient<'a> { } } HTTPClientResult { - body: None, + body: &[], + body_owned: Vec::new(), metadata: None, redirected: self.flags.redirected, fail: self.state.fail, @@ -4557,10 +4520,7 @@ impl<'a> HTTPClient<'a> { // we can ignore the body data in redirects if !self.state.flags.is_redirect_pending { if self.state.encoding.is_compressed() { - if let Some(body_out) = self.state.body_out_str { - self.state - .decompress_bytes(incoming_data, body_out::as_mut(body_out), true)?; - } + self.state.decompress_bytes(incoming_data, true)?; } else { self.state .get_body_buffer() @@ -4617,7 +4577,7 @@ impl<'a> HTTPClient<'a> { if is_done || is_streaming || content_length.is_none() { let is_final_chunk = is_done; // Move the body buffer's bytes out — process_body_buffer takes `&mut self.state` - // and may mutate `compressed_body` (via decompress_bytes' reset) or `body_out_str`, + // and may mutate `compressed_body` (via decompress_bytes' reset) or `decoded_body`, // so any `&` into `self.state` held across the call would be aliased UB. let buffer_snap = core::mem::take(&mut self.state.get_body_buffer().list); let processed = self @@ -4655,7 +4615,7 @@ impl<'a> HTTPClient<'a> { incoming_data: &[u8], ) -> crate::Result { // reshaped for borrowck — `chunked_decoder` and the body - // buffer (`compressed_body` / `body_out_str`) are disjoint fields of + // buffer (`compressed_body` / `decoded_body`) are disjoint fields of // `self.state`, so borrow them once together via the split accessor and // operate on safe references. Deep-cloning the buffer here would // diverge (mutations from process_body_buffer would be lost). @@ -4785,7 +4745,7 @@ impl<'a> HTTPClient<'a> { // Move // the bytes out so no `&` into self.state aliases the `&mut self.state` - // taken by process_body_buffer (which mutates compressed_body/body_out_str). + // taken by process_body_buffer (which mutates compressed_body/decoded_body). let buffer_snap = core::mem::take(&mut self.state.get_body_buffer().list); return self.state.process_body_buffer(buffer_snap, false); } @@ -4796,12 +4756,7 @@ impl<'a> HTTPClient<'a> { _ => { self.state.flags.received_last_chunk = true; self.handle_response_body_from_single_packet(buffer)?; - debug_assert!( - self.body_out_str() - .map(|b| b.list.as_ptr()) - .unwrap_or(core::ptr::null()) - != buffer.as_ptr() - ); + debug_assert!(self.state.decoded_body.list.as_ptr() != buffer.as_ptr()); self.report_progress(buffer.len()); Ok(true) diff --git a/src/install/NetworkTask.rs b/src/install/NetworkTask.rs index 72a2edbb388f..e51d69789fba 100644 --- a/src/install/NetworkTask.rs +++ b/src/install/NetworkTask.rs @@ -191,9 +191,9 @@ impl NetworkTask { if let Some(stream) = this.tarball_stream.as_deref_mut() { // Runs on the HTTP thread. With response-body streaming enabled, // `notify` is called once per body chunk (has_more=true) and once - // more at the end (has_more=false). `result.body` is our own - // `response_buffer`; the HTTP client reuses it for the next - // chunk, so we must consume + reset it before returning. + // more at the end (has_more=false). `result.body` borrows the + // HTTP client's scratch buffer and is cleared after this callback + // returns, so we must consume it before returning. // `metadata` is only populated on the first callback that // carries response headers. Cache the status code so both the @@ -201,9 +201,13 @@ impl NetworkTask { if let Some(m) = result.metadata.take() { stream.status_code = m.response.status_code; this.response.metadata = Some(m); + // New attempt's headers arrived — drop any bytes buffered from + // a prior failed attempt (pre-refactor `HTTPClient::start()` + // did this via `body_out_str.reset()`). + this.response_buffer.reset(); } - let chunk = this.response_buffer.list.as_slice(); + let chunk = result.body_bytes(); // Only commit to streaming extraction once we've seen a 2xx // status *and* the tarball is large enough to be worth the @@ -240,9 +244,6 @@ impl NetworkTask { // `drain()` concurrently; coercing the `&mut` to a // raw pointer here matches that contract. unsafe { TarballStream::on_chunk(stream, chunk, false, None) }; - // Hand the buffer back to the HTTP client empty so - // the next chunk starts at offset 0. - this.response_buffer.reset(); } return; } @@ -250,9 +251,8 @@ impl NetworkTask { // Final callback. If we've already started streaming, hand // over the last bytes and close; the drain task will run // once more, finish up and push to `resolve_tasks`. If not - // (whole body arrived in one go, or too small), leave - // `response_buffer` intact so the buffered extractor - // handles it. + // (whole body arrived in one go, or too small), fall through + // so the buffered extractor handles it. if committed { // SAFETY: see the `on_chunk` call above — `stream` is // live and `on_chunk` takes `*mut Self` per its @@ -280,9 +280,10 @@ impl NetworkTask { } } else if result.has_more { // Non-2xx response (or too small to stream) still - // delivering its body: accumulate in `response_buffer` - // (we did *not* reset above) so the main thread can - // inspect it. Do not enqueue until the stream ends. + // delivering its body: accumulate in `response_buffer` so + // the main thread can inspect it. Do not enqueue until the + // stream ends. + this.response_buffer.list.extend_from_slice(chunk); return; } // Fall through to the normal completion path for anything that @@ -291,6 +292,18 @@ impl NetworkTask { // streaming support. } + // Stash this callback's body bytes into our own accumulation buffer + // before `detach_lifetime` clears `result.body` to `&[]`. Covers the + // non-streaming manifest path and the tarball fall-through above. + if result.metadata.is_some() { + // First callback of a fresh attempt on the non-streaming path — + // clear stale bytes from a prior retry. The streaming fall-through + // already `.take()`d metadata and reset above, so this is a no-op + // there and accumulated chunks are preserved. + this.response_buffer.reset(); + } + result.body_into(&mut this.response_buffer.list); + // BACKREF — PackageManager owns this task and outlives it. `notify` // runs on the HTTP thread, so we never materialize a `&mut // PackageManager` here (the main thread may hold one concurrently); @@ -306,15 +319,13 @@ impl NetworkTask { unsafe { let real = async_http.real.expect("unreachable").as_ptr(); ptr::write(real, ptr::read(async_http)); - (*real).response_buffer = async_http.response_buffer; } // Preserve metadata captured on an earlier streaming callback; the // final `result` won't have it. let saved_metadata = this.response.metadata.take(); - // SAFETY: `result.body` (the only borrowed field) points at - // `this.response_buffer`, which `this` owns and outlives the stored - // `HTTPClientResult`; erase the callback-scoped `'_` to `'static` to - // match the field type. + // SAFETY: `detach_lifetime` erases the callback-scoped `'_` to + // `'static` and clears `body` to `&[]`; the body bytes were stashed + // into `this.response_buffer` above. this.response = unsafe { result.detach_lifetime() }; if this.response.metadata.is_none() { this.response.metadata = saved_metadata; @@ -635,7 +646,6 @@ impl NetworkTask { url, header_builder.entries, headers_buf, - ptr::addr_of_mut!(self.response_buffer), b"", completion_callback, http::FetchRedirect::Follow, @@ -873,7 +883,6 @@ impl NetworkTask { url, header_builder.entries, header_buf, - ptr::addr_of_mut!(self.response_buffer), b"", completion_callback, http::FetchRedirect::Follow, diff --git a/src/install/npm.rs b/src/install/npm.rs index 4f19af3c38bc..b164f015247e 100644 --- a/src/install/npm.rs +++ b/src/install/npm.rs @@ -158,14 +158,13 @@ pub fn whoami(manager: &mut PackageManager) -> Result, WhoamiError> { url, headers.entries, header_buf, - &raw mut response_buf, b"", None, None, http::FetchRedirect::Follow, ); - let res = match req.send_sync() { + let res = match req.send_sync(&mut response_buf) { Ok(res) => res, Err(bun_http::Error::Alloc(bun_alloc::AllocError)) => { return Err(WhoamiError::OutOfMemory); diff --git a/src/runtime/cli/audit_command.rs b/src/runtime/cli/audit_command.rs index e730410517a9..b24816aff445 100644 --- a/src/runtime/cli/audit_command.rs +++ b/src/runtime/cli/audit_command.rs @@ -485,13 +485,12 @@ fn send_audit_request( url, headers.entries, headers_buf, - &raw mut response_buf, &final_compressed_body, http_proxy, None, http::FetchRedirect::Follow, ); - let res = match req.send_sync() { + let res = match req.send_sync(&mut response_buf) { Ok(r) => r, Err(err) => { Output::err(err, "audit request failed", ()); diff --git a/src/runtime/cli/create_command.rs b/src/runtime/cli/create_command.rs index d133a0252743..41d1f2c42dac 100644 --- a/src/runtime/cli/create_command.rs +++ b/src/runtime/cli/create_command.rs @@ -1953,7 +1953,6 @@ impl Example { api_url, header_entries, headers_buf, - mutable, b"", http_proxy, None, @@ -1962,7 +1961,7 @@ impl Example { async_http.client.progress_node = Some(core::ptr::NonNull::from(&mut *progress)); async_http.client.flags.reject_unauthorized = env_loader.get_tls_reject_unauthorized(); - let response = async_http.send_sync()?; + let response = async_http.send_sync(mutable)?; match response.status_code() { 404 => return Err(crate::Error::GitHubRepositoryNotFound), @@ -2057,7 +2056,6 @@ impl Example { unsafe { (*URL_.get()).clone() }.unwrap(), Default::default(), b"", - mutable, b"", http_proxy, None, @@ -2066,7 +2064,7 @@ impl Example { async_http.client.progress_node = Some(core::ptr::NonNull::from(&mut *progress)); async_http.client.flags.reject_unauthorized = env_loader.get_tls_reject_unauthorized(); - let mut response = async_http.send_sync()?; + let mut response = async_http.send_sync(mutable)?; match response.status_code() { 404 => return Err(crate::Error::ExampleNotFound), @@ -2151,7 +2149,6 @@ impl Example { parsed_tarball_url, Default::default(), b"", - mutable, b"", http_proxy, None, @@ -2162,7 +2159,7 @@ impl Example { refresher.maybe_refresh(); - response = async_http.send_sync()?; + response = async_http.send_sync(mutable)?; refresher.maybe_refresh(); @@ -2197,7 +2194,6 @@ impl Example { url, Default::default(), b"", - mutable, b"", http_proxy, None, @@ -2209,7 +2205,7 @@ impl Example { async_http.client.progress_node = progress_node.map(core::ptr::NonNull::from); } - let response = match async_http.send_sync() { + let response = match async_http.send_sync(mutable) { Ok(r) => r, Err(err) => { if err.name() == "EAGAIN" { diff --git a/src/runtime/cli/pm_view_command.rs b/src/runtime/cli/pm_view_command.rs index 4ebb0b698242..2e25269bf962 100644 --- a/src/runtime/cli/pm_view_command.rs +++ b/src/runtime/cli/pm_view_command.rs @@ -118,7 +118,6 @@ pub(crate) fn view( url, headers.entries, header_buf, - &raw mut response_buf, b"", http_proxy, None, @@ -126,7 +125,7 @@ pub(crate) fn view( ); req.client.flags.reject_unauthorized = manager.tls_reject_unauthorized(); - let res = match req.send_sync() { + let res = match req.send_sync(&mut response_buf) { Ok(r) => r, Err(err) => { Output::err(err, "view request failed to send", ()); diff --git a/src/runtime/cli/publish_command.rs b/src/runtime/cli/publish_command.rs index 0d57238f4292..1cc25d83fd38 100644 --- a/src/runtime/cli/publish_command.rs +++ b/src/runtime/cli/publish_command.rs @@ -838,14 +838,13 @@ impl PublishCommand { package_url, headers.entries, headers.content.written_slice(), - &raw mut response_buf, b"", None, None, http::FetchRedirect::Follow, ); - let Ok(res) = req.send_sync() else { + let Ok(res) = req.send_sync(&mut response_buf) else { return false; }; if res.status_code() != 200 { @@ -964,14 +963,13 @@ impl PublishCommand { publish_url.clone(), publish_headers.entries, publish_headers.content.written_slice(), - &raw mut response_buf, publish_req_body, None, None, http::FetchRedirect::Follow, ); - let res = match req.send_sync() { + let res = match req.send_sync(&mut response_buf) { Ok(r) => r, Err(e) => { if e == bun_http::Error::Alloc(bun_alloc::AllocError) { @@ -1059,14 +1057,13 @@ impl PublishCommand { publish_url, otp_headers.entries, otp_headers.content.written_slice(), - &raw mut response_buf, publish_req_body, None, None, http::FetchRedirect::Follow, ); - let otp_res = match otp_req.send_sync() { + let otp_res = match otp_req.send_sync(&mut response_buf) { Ok(r) => r, Err(e) => { if e == bun_http::Error::Alloc(bun_alloc::AllocError) { @@ -1297,14 +1294,13 @@ impl PublishCommand { done_url.clone(), auth_headers.entries.clone()?, auth_headers.content.written_slice(), - response_buf, b"", None, None, http::FetchRedirect::Follow, ); - let res = match req.send_sync() { + let res = match req.send_sync(response_buf) { Ok(r) => r, Err(e) => { if e == bun_http::Error::Alloc(bun_alloc::AllocError) { diff --git a/src/runtime/cli/run_command.rs b/src/runtime/cli/run_command.rs index b6983cb2c711..b9de25b12d61 100644 --- a/src/runtime/cli/run_command.rs +++ b/src/runtime/cli/run_command.rs @@ -3149,7 +3149,7 @@ impl RemoteImageDownload { fn on_done( this: *mut RemoteImageDownload, async_http: *mut bun_http::AsyncHTTP<'static>, - _result: bun_http::HTTPClientResult<'_>, + mut result: bun_http::HTTPClientResult<'_>, ) { // The worker's // ThreadlocalAsyncHTTP is about to be freed, so copy its @@ -3167,8 +3167,8 @@ impl RemoteImageDownload { // `*real.as_ptr() = …` would run Drop on the previous // `this.async_http` (whose state the fresh copy still aliases). real.as_ptr().write(::core::ptr::read(async_http)); - (*real.as_ptr()).response_buffer = async_http.response_buffer; } + result.body_into(&mut this.response_buffer.list); // Channel payload is a placeholder tick — the main thread // walks `downloads[]` to read per-task state after N wakeups. let _ = (*this.done).write_item(0); @@ -3297,17 +3297,12 @@ impl RunCommand { let url = &*::core::ptr::addr_of!((*slot).url); ::core::slice::from_raw_parts(url.as_ptr(), url.len()) }; - // SAFETY: `slot` is the freshly-allocated `MaybeUninit` heap slot - // and `response_buffer` was `ptr::write`n above; address is valid. - let response_buffer_ptr: *mut bun_core::MutableString = - unsafe { ::core::ptr::addr_of_mut!((*slot).response_buffer) }; let d_ptr: *mut RemoteImageDownload = slot; let async_http = bun_http::AsyncHTTP::init( bun_http::Method::GET, bun_url::URL::parse(url_static), Default::default(), b"", - response_buffer_ptr, b"", bun_http::HTTPClientResultCallback::new::( d_ptr, diff --git a/src/runtime/cli/upgrade_command.rs b/src/runtime/cli/upgrade_command.rs index 4d193ad436e9..fcd16bf1b0bf 100644 --- a/src/runtime/cli/upgrade_command.rs +++ b/src/runtime/cli/upgrade_command.rs @@ -264,7 +264,6 @@ impl UpgradeCommand { api_url, header_entries, headers_buf, - std::ptr::from_mut::(metadata_body), b"", http_proxy, None, @@ -278,7 +277,7 @@ impl UpgradeCommand { // frame returns, so the pointee outlives every use. async_http.client.progress_node = Some(NonNull::from(progress.as_deref_mut().unwrap())); } - let response = async_http.send_sync()?; + let response = async_http.send_sync(metadata_body)?; match response.status_code() { 404 => return Err(crate::Error::HTTP404), @@ -657,7 +656,6 @@ impl UpgradeCommand { zip_url, headers::EntryList::default(), b"", - std::ptr::from_mut::(zip_file_buffer), b"", http_proxy, None, @@ -669,7 +667,7 @@ impl UpgradeCommand { Some(NonNull::new(progress).expect("leaked Box is non-null")); async_http.client.flags.reject_unauthorized = env_loader.get_tls_reject_unauthorized(); - let response = async_http.send_sync()?; + let response = async_http.send_sync(zip_file_buffer)?; match response.status_code() { 404 => { diff --git a/src/runtime/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index 727e02fb664c..8b65504edb06 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -50,6 +50,9 @@ impl Taskable for FetchTasklet { bun_output::declare_scope!(FetchTasklet, visible); +/// Upper bound on the Content-Length-driven `reserve_exact` in `callback()`. +const SCHEDULED_PRERESERVE_MAX: usize = 256 * 1024 * 1024; + use http::signals::BodyReceiveMode; #[derive(bun_ptr::ThreadSafeRefCounted)] @@ -72,8 +75,6 @@ pub struct FetchTasklet { // borrowed for `acquire/release`. Model as a raw pointer. pub(crate) request_body_streaming_buffer: Option>, - /// buffer being used by AsyncHTTP - pub(crate) response_buffer: MutableString, /// buffer used to stream response to JS pub(crate) scheduled_response_buffer: MutableString, /// response weak ref we need this to track the response JS lifetime @@ -115,6 +116,12 @@ pub struct FetchTasklet { // Custom Hostname pub(crate) hostname: Option>, pub(crate) is_waiting_body: bool, + /// Set by `on_start_buffering_callback` (JS thread) and read by + /// `callback()` (HTTP thread, under `mutex`): the body is being + /// accumulated in `scheduled_response_buffer` for a buffered consumer. + /// Distinguishes that path from `drop_backpressure_if_unobserved`, which + /// also sets `BufferAll` but still delivers per chunk. + pub(crate) is_buffering_body: AtomicBool, pub(crate) is_waiting_abort: bool, pub(crate) is_waiting_request_stream_start: bool, pub(crate) mutex: Mutex, @@ -465,7 +472,6 @@ impl FetchTasklet { drop(metadata); } - self.response_buffer = MutableString::default(); self.response.clear(); if let Some(response) = self.native_response.take() { // SAFETY: `response` is the +1 ref held in `native_response`. @@ -798,7 +804,12 @@ impl FetchTasklet { if buffer_reset.get() { // SAFETY: `self` outlives this defer (it's a local in this fn) and no other // borrow of scheduled_response_buffer is live at scope exit / `?` unwind. - unsafe { (*scheduled_buf).reset() }; + let list = unsafe { &mut (*scheduled_buf).list }; + if list.capacity() > http::DECODED_BODY_RETAIN_CAP { + *list = Vec::new(); + } else { + list.clear(); + } } } @@ -1703,6 +1714,9 @@ impl FetchTasklet { ) { let this = Self::from_ctx(ctx); this.readable_stream_ref = ReadableStreamStrong::init(readable, global_this); + // A ByteStream now drains scheduled_response_buffer per chunk; undo any + // buffered-consumer reservation request so callback() stops growing it. + this.is_buffering_body.store(false, Ordering::Release); } fn on_start_streaming_http_response_body_callback(ctx: *mut c_void) -> DrainResult { @@ -1735,6 +1749,10 @@ impl FetchTasklet { } this.mutex.lock(); + // A ByteStream is attaching; clear the buffered-consumer reserve gate + // under the mutex so the HTTP thread cannot observe the stale `true` + // in `callback()` between this unlock and `on_readable_stream_available`. + this.is_buffering_body.store(false, Ordering::Release); // explicit unlock at each return // (no `?` paths between lock and unlock, so a guard is unnecessary). let size_hint = this.get_size_hint(); @@ -1805,6 +1823,7 @@ impl FetchTasklet { fn on_start_buffering_callback(ctx: *mut c_void) { let this = Self::from_ctx(ctx); this.poll_ref.ref_(bun_io::js_vm_ctx()); + this.is_buffering_body.store(true, Ordering::Release); if this .signal_store .set_receive_mode_terminal(BodyReceiveMode::BufferAll) @@ -2011,7 +2030,6 @@ impl FetchTasklet { global_this: GlobalRef::from(global_this), request_body: fetch_options.body, request_body_streaming_buffer: None, - response_buffer: MutableString::default(), scheduled_response_buffer: MutableString::default(), response: jsc::Weak::default(), native_response: None, @@ -2032,6 +2050,7 @@ impl FetchTasklet { upgraded_connection: fetch_options.upgraded_connection, hostname: fetch_options.hostname, is_waiting_body: false, + is_buffering_body: AtomicBool::new(false), is_waiting_abort: false, is_waiting_request_stream_start: false, mutex: Mutex::new(), @@ -2097,9 +2116,8 @@ impl FetchTasklet { // `heap::alloc`, so erase the borrow lifetimes through raw pointers. // SAFETY: `fetch_tasklet_ptr` is a stable heap allocation that outlives // the AsyncHTTP (dropped together in `deinit`); the slices below borrow - // its `request_headers.buf`, `request_body`, `hostname`, and - // `response_buffer` fields which are not reallocated for the lifetime - // of the request. + // its `request_headers.buf`, `request_body`, and `hostname` fields + // which are not reallocated for the lifetime of the request. // SAFETY (`Interned::assume` — Population B, holder-backed): // `fetch_tasklet_ptr` is a `heap::alloc`'d `FetchTasklet` whose // `request_headers.buf` / `request_body` / `hostname` fields are not @@ -2119,7 +2137,6 @@ impl FetchTasklet { .as_deref() // SAFETY: see block note above — same `FetchTasklet` owner. .map(|s| unsafe { bun_ptr::Interned::assume(s) }.as_bytes()); - let response_buffer: *mut MutableString = &raw mut fetch_tasklet.response_buffer; // `MultiArrayList` owns its // allocation, so clone; AsyncHTTP::init clones again for the client. let header_entries = bun_core::handle_oom(fetch_tasklet.request_headers.entries.clone()); @@ -2132,7 +2149,6 @@ impl FetchTasklet { url, header_entries, headers_buf, - response_buffer, request_body_slice, // handles response events (on headers, on body, etc.) http::HTTPClientResultCallback::new_with_release::( @@ -2500,7 +2516,7 @@ impl FetchTasklet { fn callback( task: *mut FetchTasklet, async_http: *mut AsyncHTTP<'static>, - result: HTTPClientResult, + mut result: HTTPClientResult, ) { // at this point only this thread is accessing result to is no race condition let is_done = !result.has_more; @@ -2528,25 +2544,20 @@ impl FetchTasklet { result.is_success(), task_ref.signal_store.body_receive_mode(), result.has_more, - result.body.as_ref().map(|b| b.list.len()).unwrap_or(0) - ); - - // Verify the aliasing invariant (see comment below). - debug_assert!( - result - .body - .as_deref() - .is_none_or(|b| core::ptr::eq(b, &raw const task_ref.response_buffer)), - "HTTPClientResult.body must alias FetchTasklet.response_buffer", + result.body.len() ); let prev_metadata = task_ref.result.metadata.take(); let prev_cert_info = task_ref.result.certificate_info.take(); let prev_can_stream = task_ref.result.can_stream; - // SAFETY: lifetime erasure — `HTTPClientResult<'a>` borrows the - // `*mut MutableString` we passed into `AsyncHTTP::init` (which lives - // in `self.response_buffer` for the FetchTasklet's lifetime); widen - // `'_` → `'static` to store it. + // `result.body` borrows the HTTP thread's scratch buffer on non-terminal + // callbacks; the terminal callback carries the bytes in `body_owned` + // instead. Capture both before `detach_lifetime` clears them in the + // stored copy. + let body: &[u8] = result.body; + let body_owned: Vec = core::mem::take(&mut result.body_owned); + // SAFETY: lifetime erasure for non-body fields; `body` is stored as + // `&'static []` so no borrow escapes. task_ref.result = unsafe { result.detach_lifetime() }; // can_stream is a one-shot signal to start the request body stream; don't let a // later coalesced result clobber it before the JS thread sees it. @@ -2572,15 +2583,8 @@ impl FetchTasklet { task_ref.body_size = task_ref.result.body_size; let success = task_ref.result.is_success(); - // `result.body` always aliases `task_ref.response_buffer` (the - // `*mut MutableString` passed to `AsyncHTTP::init` at FetchTasklet::create flows - // through `HTTPClient.state.body_out_str` and back out in the result). Asserted - // above before the lifetime-erasing assignment; the bytes are already in place, so - // no copy is needed and the `reset()` calls below operate on the right allocation. if task_ref.signal_store.body_receive_mode() == BodyReceiveMode::Ignore { - task_ref.response_buffer.reset(); - if task_ref.scheduled_response_buffer.list.capacity() > 0 { task_ref.scheduled_response_buffer = MutableString::default(); } @@ -2593,22 +2597,42 @@ impl FetchTasklet { } return; } - } else { - if success { - bun_core::handle_oom( - task_ref - .scheduled_response_buffer - .write(task_ref.response_buffer.list.as_slice()), - ); - if task_ref.result.has_more && !task_ref.scheduled_response_buffer.list.is_empty() { - let _ = task_ref.signal_store.try_transition_receive_mode( - BodyReceiveMode::AutoPause, - BodyReceiveMode::Paused, - ); + } else if success { + let scheduled = &mut task_ref.scheduled_response_buffer; + if body.is_empty() && !body_owned.is_empty() && scheduled.list.is_empty() { + scheduled.list = body_owned; + } else { + // Grow to Content-Length once so the per-packet append below + // doesn't leave the ~2x doubling over-capacity that the + // ArrayBuffer would adopt. Gated on `is_buffering_body` + // (set by `on_start_buffering_callback`), not the raw + // `BufferAll` mode: `drop_backpressure_if_unobserved` also + // sets `BufferAll` while still draining per chunk. + if task_ref.is_buffering_body.load(Ordering::Acquire) { + if let http::BodySize::ContentLength(n) = task_ref.body_size { + if n > scheduled.list.capacity() { + let additional = n + .min(SCHEDULED_PRERESERVE_MAX) + .saturating_sub(scheduled.list.len()); + let _ = scheduled.list.try_reserve_exact(additional); + } + } + } + let chunk = if body.is_empty() { + body_owned.as_slice() + } else { + body + }; + if !chunk.is_empty() { + bun_core::handle_oom(scheduled.write(chunk)); } } - // reset for reuse - task_ref.response_buffer.reset(); + if task_ref.result.has_more && !task_ref.scheduled_response_buffer.list.is_empty() { + let _ = task_ref.signal_store.try_transition_receive_mode( + BodyReceiveMode::AutoPause, + BodyReceiveMode::Paused, + ); + } } if let Err(has_schedule_callback) = task_ref.has_schedule_callback.compare_exchange( diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs index c9e4a6d4a821..b86d2b8c41e0 100644 --- a/src/runtime/webcore/s3/client.rs +++ b/src/runtime/webcore/s3/client.rs @@ -350,7 +350,6 @@ pub(crate) fn list_objects( url, task.headers.entries.clone().expect("OOM"), headers_buf, - &raw mut task.response_buffer, b"", bun_http::HTTPClientResultCallback::new::( task_ptr, @@ -1233,7 +1232,6 @@ fn download_stream( signal_store: Default::default(), signals: Default::default(), poll_ref: bun_io::KeepAlive::init(), - response_buffer: MutableString::default(), mutex: Default::default(), request_error: None, reported_response_buffer: MutableString::default(), @@ -1283,7 +1281,6 @@ fn download_stream( url, task.headers.entries.clone().expect("OOM"), headers_buf, - &raw mut task.response_buffer, b"", bun_http::HTTPClientResultCallback::new::( task_ptr, diff --git a/src/runtime/webcore/s3/download_stream.rs b/src/runtime/webcore/s3/download_stream.rs index 2dcf0be2443b..0e5553fc566d 100644 --- a/src/runtime/webcore/s3/download_stream.rs +++ b/src/runtime/webcore/s3/download_stream.rs @@ -30,7 +30,6 @@ pub struct S3HttpDownloadStreamingTask { pub(crate) signals: Signals, pub poll_ref: KeepAlive, - pub(crate) response_buffer: MutableString, pub(crate) mutex: Mutex, pub(crate) reported_response_buffer: MutableString, /// The HTTP-level failure, if any. Guarded by `mutex`; the `request_error` @@ -70,7 +69,6 @@ impl Default for S3HttpDownloadStreamingTask { signal_store: bun_http::signals::Store::default(), signals: Signals::default(), poll_ref: KeepAlive::default(), - response_buffer: MutableString::default(), mutex: Mutex::default(), reported_response_buffer: MutableString::default(), request_error: None, @@ -262,7 +260,7 @@ impl S3HttpDownloadStreamingTask { fn process_http_callback( &mut self, async_http: &mut AsyncHTTP<'static>, - result: HTTPClientResult, + mut result: HTTPClientResult, ) -> bool { // lets lock and unlock to be safe we know the state is not in the middle of a callback when locked // The RAII guard unlocks on every @@ -288,22 +286,9 @@ impl S3HttpDownloadStreamingTask { should_enqueue ); + result.body_into(&mut self.reported_response_buffer.list); if should_enqueue { - if let Some(body) = result.body { - // `body` is `&this.response_buffer`, so a `ptr::read` + assign here would - // run Drop on the old `self.response_buffer`, freeing the Vec allocation - // that `body` (and the freshly-stored value) still point at — a - // use-after-free / double-free. Instead, append `body`'s bytes to - // `reported_response_buffer`, then reset the buffer, operating on `body` - // directly. - if !body.list.as_slice().is_empty() { - let _ = self.reported_response_buffer.write(body.list.as_slice()); - } - body.reset(); - if self.reported_response_buffer.list.as_slice().is_empty() && !is_done { - return false; - } - } else if !is_done { + if self.reported_response_buffer.list.is_empty() && !is_done { return false; } if let Err(has_schedule_callback) = self.has_schedule_callback.compare_exchange( @@ -363,7 +348,7 @@ impl Drop for S3HttpDownloadStreamingTask { self.poll_ref.unref(bun_io::posix_event_loop::get_vm_ctx( bun_io::AllocatorType::Js, )); - // response_buffer, reported_response_buffer, headers, sign_result, range, proxy_url: + // reported_response_buffer, headers, sign_result, range, proxy_url: // dropped automatically (Box/Vec-backed fields). // SAFETY: `http` is always initialised before the task is scheduled / dropped. let http = unsafe { self.http.assume_init_mut() }; diff --git a/src/runtime/webcore/s3/simple_request.rs b/src/runtime/webcore/s3/simple_request.rs index 72befdfe0937..eff6aa931a81 100644 --- a/src/runtime/webcore/s3/simple_request.rs +++ b/src/runtime/webcore/s3/simple_request.rs @@ -123,8 +123,6 @@ pub struct S3HttpSimpleTask { pub(crate) callback_context: *mut c_void, pub callback: Callback, pub(crate) response_buffer: MutableString, - // `'static` here because `result.body` (when set) points at our own - // `response_buffer` — self-referential, so the borrow lives as long as the task. pub(crate) result: HTTPClientResult<'static>, pub(crate) concurrent_task: ConcurrentTask, /// Owned dupe of the proxy URL. The env-derived proxy slice can be freed @@ -238,8 +236,8 @@ impl S3HttpSimpleTask { if let Some(err) = self.result.fail { code = err.name().as_bytes(); has_error_code = true; - } else if let Some(body) = &self.result.body { - let bytes = body.list.as_slice(); + } else { + let bytes = self.response_buffer.list.as_slice(); if !bytes.is_empty() { message = bytes; if let Some(start) = strings::index_of(bytes, b"") { @@ -281,8 +279,8 @@ impl S3HttpSimpleTask { if let Some(err) = self.result.fail { code = err.name().as_bytes(); - } else if let Some(body) = &self.result.body { - let bytes = body.list.as_slice(); + } else { + let bytes = self.response_buffer.list.as_slice(); let mut has_error = false; if !bytes.is_empty() { message = bytes; @@ -309,8 +307,6 @@ impl S3HttpSimpleTask { if (!has_error && status == 200) || status == 206 { return Ok(false); } - } else if status == 200 || status == 206 { - return Ok(false); } self.callback.fail(code, message, self.callback_context)?; Ok(true) @@ -365,18 +361,15 @@ impl S3HttpSimpleTask { }, Callback::ListObjects(callback) => match response.status_code { 200 => { - if let Some(body) = &this.result.body { - // parse_s3_list_objects_result is infallible (alloc-only - // failure modes abort). - let success = - list_objects::parse_s3_list_objects_result(body.list.as_slice()); - callback( - S3ListObjectsResult::Success(Box::new(success)), - this.callback_context, - )?; - } else { - this.error_with_body(ErrorType::Failure)?; - } + // parse_s3_list_objects_result is infallible (alloc-only + // failure modes abort). + let success = list_objects::parse_s3_list_objects_result( + this.response_buffer.list.as_slice(), + ); + callback( + S3ListObjectsResult::Success(Box::new(success)), + this.callback_context, + )?; } 404 => this.error_with_body(ErrorType::NotFound)?, _ => this.error_with_body(ErrorType::Failure)?, @@ -434,7 +427,7 @@ impl S3HttpSimpleTask { pub(crate) fn http_callback( this: *mut Self, async_http: *mut AsyncHTTP<'static>, - result: HTTPClientResult<'_>, + mut result: HTTPClientResult<'_>, ) { // SAFETY: `this` was produced by `S3HttpSimpleTask::new` and is exclusively owned by the // HTTP thread until enqueued back to the JS thread below. @@ -444,8 +437,9 @@ impl S3HttpSimpleTask { // A close-delimited body (no Content-Length, no Transfer-Encoding) reports progress again // at EOF with `metadata: None`, so carry the earlier one across the assignment below. let previous_metadata = this.result.metadata.take(); - // SAFETY: `result.body` (the only borrowed field) points at `this.response_buffer`, which - // lives for the task's lifetime — extending to `'static` here is sound for self-reference. + result.body_into(&mut this.response_buffer.list); + // SAFETY: `body` is the only lifetime-carrying field and `body_into` + // just consumed it; the stored `&'static []` is never read. this.result = unsafe { result.detach_lifetime() }; if this.result.metadata.is_none() { this.result.metadata = previous_metadata; @@ -461,10 +455,6 @@ impl S3HttpSimpleTask { // SAFETY: `async_http` is a valid live pointer for the duration of this callback; // `this.http` was previously initialised in `execute_simple_s3_request`. unsafe { core::ptr::write(this.http.as_mut_ptr(), core::ptr::read(async_http)) }; - // `async_http.response_buffer == &this.response_buffer`, so copying it back would be - // a self-assignment: the `=` would drop the live Vec before re-installing a stale - // bitwise duplicate (UAF + double-free), so we simply omit it — - // `this.response_buffer` already holds the body. if is_done { // compute the raw self-pointer before borrowing `this.concurrent_task` // to avoid a stacked-borrows / aliasing diagnostic on `*this`. @@ -669,7 +659,6 @@ pub(crate) fn execute_simple_s3_request( url, task.headers.entries.clone().expect("OOM"), headers_buf, - &raw mut task.response_buffer, body, HTTPClientResultCallback::new::( task_ptr, diff --git a/src/standalone_graph/StandaloneModuleGraph.rs b/src/standalone_graph/StandaloneModuleGraph.rs index 821015186f5f..e4ccd2c5364d 100644 --- a/src/standalone_graph/StandaloneModuleGraph.rs +++ b/src/standalone_graph/StandaloneModuleGraph.rs @@ -1692,7 +1692,6 @@ pub(crate) fn download_to_path( url, Default::default(), b"", - &raw mut *compressed_archive_bytes, b"", http_proxy, None, @@ -1701,7 +1700,7 @@ pub(crate) fn download_to_path( async_http.client.progress_node = core::ptr::NonNull::new(core::ptr::from_mut(progress)); async_http.client.flags.reject_unauthorized = reject_unauthorized; - let send_result = async_http.send_sync(); + let send_result = async_http.send_sync(&mut compressed_archive_bytes); progress.end(); let status_code = send_result?.status_code() as u16; diff --git a/test/js/web/fetch/fetch-buffer-peak-fixture.ts b/test/js/web/fetch/fetch-buffer-peak-fixture.ts new file mode 100644 index 000000000000..2cb14db4ed00 --- /dev/null +++ b/test/js/web/fetch/fetch-buffer-peak-fixture.ts @@ -0,0 +1,42 @@ +// Measures the client-side resident set across a single large buffered +// fetch().arrayBuffer(). The server lives in the parent test process so this +// process's RSS reflects only the fetch side. +// +// Uses process.memoryUsage.rss() rather than resourceUsage().maxRSS: on Linux +// ru_maxrss survives exec, so a child spawned from a heavy test runner +// inherits the parent's RSS as its initial high-water mark. +// +// Under ASAN's default quarantine (debug builds) or mimalloc's page cache +// (release) the intermediate reallocations from amortized-doubling growth stay +// resident past the fetch, so the post-arrayBuffer RSS reflects the total +// allocation volume rather than just the final live buffer. +// +// Prints one JSON line: { bodyMB, rssBeforeMB, rssAfterMB } + +const url = process.env.SERVER!; +const bodyBytes = Number(process.env.BODY_BYTES!); + +Bun.gc(true); +const rssBefore = process.memoryUsage.rss(); + +const res = await fetch(url); +const buf = await res.arrayBuffer(); + +const rssAfter = process.memoryUsage.rss(); + +if (buf.byteLength !== bodyBytes) { + throw new Error(`expected ${bodyBytes} bytes, got ${buf.byteLength}`); +} + +// Keep both referenced past the measurement so neither is collected early. +if (!res.ok) throw new Error("unreachable"); +if (buf.byteLength === 0) throw new Error("unreachable"); + +const mb = (n: number) => Math.round(n / 1024 / 1024); +console.log( + JSON.stringify({ + bodyMB: mb(bodyBytes), + rssBeforeMB: mb(rssBefore), + rssAfterMB: mb(rssAfter), + }), +); diff --git a/test/js/web/fetch/fetch-leak.test.ts b/test/js/web/fetch/fetch-leak.test.ts index 0f903061168c..5c9a9cf290d1 100644 --- a/test/js/web/fetch/fetch-leak.test.ts +++ b/test/js/web/fetch/fetch-leak.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tls as COMMON_CERT, gc, isASAN, isCI, isDebug } from "harness"; import { once } from "node:events"; import { createServer } from "node:http"; +import net from "node:net"; import { join } from "node:path"; describe("fetch doesn't leak", () => { @@ -962,3 +963,66 @@ test("aborting in-flight streaming fetch() responses does not retain the buffere expect(stderr).not.toContain("LEAK"); expect(exitCode).toBe(0); }); + +test("fetch().arrayBuffer() of a large Content-Length body peaks at ~1x the payload", async () => { + // The per-packet handoff in FetchTasklet::callback appended each socket read + // into scheduled_response_buffer via extend_from_slice, so the Vec grew by + // amortized doubling. For a body just past a doubling step that left the + // allocation the ArrayBuffer adopts at ~2x the payload, and the intermediate + // reallocations spiked ru_maxrss well past 2x. Reserving Content-Length + // exactly up front keeps it to a single allocation that is moved through to + // the ArrayBuffer. + // + // Body size is 128 MiB + 1 MiB so the old doubling growth would have crossed + // the 128 -> 256 step, making the unfixed peak reliably > 2x body. + const bodyBytes = 129 * 1024 * 1024; + const chunk = Buffer.alloc(256 * 1024, "abcdefghij"); + const server = net.createServer(socket => { + socket.once("data", () => { + socket.write(`HTTP/1.1 200 OK\r\nContent-Length: ${bodyBytes}\r\nConnection: close\r\n\r\n`); + let sent = 0; + const pump = () => { + while (sent < bodyBytes) { + const n = Math.min(chunk.length, bodyBytes - sent); + sent += n; + if (!socket.write(n === chunk.length ? chunk : chunk.subarray(0, n))) { + socket.once("drain", pump); + return; + } + } + socket.end(); + }; + pump(); + }); + socket.on("error", () => {}); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address(); + + try { + await using proc = Bun.spawn({ + cmd: [bunExe(), "--smol", join(import.meta.dir, "fetch-buffer-peak-fixture.ts")], + env: { + ...bunEnv, + SERVER: `http://127.0.0.1:${port}/`, + BODY_BYTES: String(bodyBytes), + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + console.log(stdout.trim()); + expect(stderr).toBe(""); + const { bodyMB, rssBeforeMB, rssAfterMB } = JSON.parse(stdout.trim()); + expect(bodyMB).toBe(129); + // Unfixed: the doubling reallocations (retained by ASAN quarantine / mimalloc + // page cache) leave RSS at >= 2x body over the pre-fetch resident + // set (release linux ~2.9x, debug+ASAN default quarantine ~2.7x). + // Fixed: a single exact allocation, ~1.0x body. + const delta = rssAfterMB - rssBeforeMB; + expect(delta).toBeLessThan(1.5 * bodyMB); + expect(exitCode).toBe(0); + } finally { + server.close(); + } +}, 60000);