diff --git a/src/http/H2Client.rs b/src/http/H2Client.rs index e25bb2bd34d4..5a275b2f5b74 100644 --- a/src/http/H2Client.rs +++ b/src/http/H2Client.rs @@ -106,8 +106,8 @@ pub(crate) mod bridge { self.do_redirect::(ctx, socket); } #[inline] - pub fn h2_clone_metadata(&mut self) { - self.clone_metadata(); + pub fn h2_clone_metadata(&mut self, response: &bun_picohttp::Response<'_>) { + self.clone_metadata(response); } #[inline] pub fn h2_handle_response_body( diff --git a/src/http/InternalState.rs b/src/http/InternalState.rs index 09e70ea60ee3..52e3c20a8f96 100644 --- a/src/http/InternalState.rs +++ b/src/http/InternalState.rs @@ -13,11 +13,6 @@ bun_core::define_scoped_log!(log, HTTPInternalState, hidden); pub struct InternalState<'a> { pub response_message_buffer: MutableString, - /// pending response is the temporary storage for the response headers, url and status code - /// this uses shared_response_headers_buf to store the headers - /// this will be turned None once the metadata is cloned - pub pending_response: Option>, - /// 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 @@ -121,7 +116,6 @@ impl Default for InternalState<'_> { fn default() -> Self { Self { response_message_buffer: MutableString::init_empty(), - pending_response: None, cloned_metadata: None, flags: InternalStateFlags::new(), transfer_encoding: Encoding::Identity, @@ -157,7 +151,6 @@ impl<'a> InternalState<'a> { response_message_buffer: MutableString::init_empty(), body_out_str: Some(NonNull::from(body_out_str)), stage: Stage::Pending, - pending_response: None, ..Default::default() } } diff --git a/src/http/ProxyTunnel.rs b/src/http/ProxyTunnel.rs index c609d3563fed..698ce01f3bae 100644 --- a/src/http/ProxyTunnel.rs +++ b/src/http/ProxyTunnel.rs @@ -290,7 +290,6 @@ fn on_data(ctx: *mut HTTPClient, decoded_data: &[u8]) { // arriving here is unexpected. if this.state.flags.is_waiting_for_cert_check { scoped_log!(http_proxy_tunnel, "ProxyTunnel onData while parked"); - this.state.pending_response = None; // SAFETY: `this` dead (NLL); reenter via raw ptr. ProxyTunnel::close_from_callback(proxy_nn, crate::Error::UnexpectedData); return; @@ -354,7 +353,6 @@ fn on_data(ctx: *mut HTTPClient, decoded_data: &[u8]) { } _ => { scoped_log!(http_proxy_tunnel, "ProxyTunnel onData unexpected data"); - this.state.pending_response = None; // SAFETY: `this` dead (NLL); reenter via raw ptr. ProxyTunnel::close_from_callback(proxy_nn, crate::Error::UnexpectedData); } diff --git a/src/http/h2_client/ClientSession.rs b/src/http/h2_client/ClientSession.rs index 3418b3dad723..cf29b96945c2 100644 --- a/src/http/h2_client/ClientSession.rs +++ b/src/http/h2_client/ClientSession.rs @@ -1007,7 +1007,9 @@ impl ClientSession { if stream.headers_ready { stream.headers_ready = false; - let result = match self.apply_headers(stream, client) { + let (result, response) = match client + .apply_multiplexed_headers(stream.status_code, &stream.decoded_headers) + { Ok(r) => r, Err(err) => { self.rst_stream(stream, wire::ErrorCode::CANCEL); @@ -1034,12 +1036,14 @@ impl ClientSession { client.h2_do_redirect(self.ctx, self.socket); return true; } + // Deep-copy before detaching: `response` borrows + // `stream.decoded_headers`. + client.h2_clone_metadata(&response); if result == HeaderResult::Finished || (stream.remote_closed() && stream.body_buffer.is_empty()) { stream.client = None; client.h2 = None; - client.h2_clone_metadata(); client.state.flags.received_last_chunk = true; // .finished = HEAD/204/304: no body is expected regardless of // any Content-Length header, so clear it. Otherwise leave the @@ -1050,7 +1054,6 @@ impl ClientSession { } return self.finish_stream(stream, client); } - client.h2_clone_metadata(); // Mirror the h1 path: deliver headers // to JS now so `await fetch()` resolves and `getReader()` can enable // response_body_streaming. Without this, a content-length response @@ -1133,19 +1136,6 @@ impl ClientSession { client.h2_progress_update(self.ctx, self.socket); true } - - /// Hand the pre-decoded response headers to the existing HTTP/1.1 - /// metadata pipeline (`handleResponseMetadata` + `cloneMetadata`). - fn apply_headers( - &mut self, - stream: &mut Stream, - client: &mut HTTPClient, - ) -> Result { - // SAFETY: decoded_headers borrow stream.decoded_bytes, which outlives - // the synchronous clone_metadata that follows in `process_stream` — - // see `HTTPClient::apply_multiplexed_headers` contract. - client.apply_multiplexed_headers(stream.status_code, &stream.decoded_headers) - } } impl Drop for ClientSession { diff --git a/src/http/h3_client/ClientSession.rs b/src/http/h3_client/ClientSession.rs index e8ad71ca5ef5..59f32f3adab4 100644 --- a/src/http/h3_client/ClientSession.rs +++ b/src/http/h3_client/ClientSession.rs @@ -297,7 +297,9 @@ impl ClientSession { if st.status_code != 0 && !st.headers_delivered { st.headers_delivered = true; - let result = match apply_headers(st, client) { + let (result, response) = match client + .apply_multiplexed_headers(u32::from(st.status_code), &st.decoded_headers) + { Ok(r) => r, Err(e) => return self.fail(stream, e), }; @@ -308,7 +310,7 @@ impl ClientSession { let client = client_mut(client_ptr); return client.do_redirect_h3(); } - client.clone_metadata(); + client.clone_metadata(&response); client.state.flags.received_last_chunk = true; if result == HeaderResult::Finished { client.state.content_length = Some(0); @@ -318,7 +320,7 @@ impl ClientSession { let client = client_mut(client_ptr); return finish(client); } - client.clone_metadata(); + client.clone_metadata(&response); if client.signals.get(Signal::HeaderProgress) { client.progress_update_h3(); } @@ -465,13 +467,6 @@ pub(super) fn session_mut<'a>(p: *mut ClientSession) -> &'a mut ClientSession { unsafe { &mut *p } } -fn apply_headers(stream: &mut Stream, client: &mut HTTPClient) -> crate::Result { - // SAFETY: decoded_headers borrow the lsquic hset, which is deep-copied by - // `clone_metadata` inside the same lsquic callback before lsquic frees it - // — see `HTTPClient::apply_multiplexed_headers` contract. - client.apply_multiplexed_headers(u32::from(stream.status_code), &stream.decoded_headers) -} - fn finish(client: &mut HTTPClient) { if let Some(cl) = client.state.content_length { if client.state.total_body_received != cl { diff --git a/src/http/lib.rs b/src/http/lib.rs index 7288f064ab96..eca6a662c26a 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -359,17 +359,16 @@ impl HTTPClient<'_> { /// undo the HTTP/1.1-specific framing decisions that don't apply when the /// transport delimits the body (h2 DATA frames / h3 STREAM frames). /// - /// SAFETY CONTRACT: `headers` borrows caller-owned storage - /// (`stream.decoded_bytes` for h2, the lsquic hset for h3) that is - /// lifetime-erased into `state.pending_response`. The caller MUST invoke - /// `clone_metadata` (which deep-copies the header bytes) synchronously - /// before that backing storage is freed. Both call sites already do. + /// Returns the (possibly-mutated) response so the caller can pass it to + /// `clone_metadata` once the redirect decision has been made; the borrow of + /// `headers` flows through, so the deep copy is checked by the compiler + /// rather than by a call-ordering contract. #[inline] - pub(crate) fn apply_multiplexed_headers( + pub(crate) fn apply_multiplexed_headers<'h>( &mut self, status_code: u32, - headers: &[picohttp::Header], - ) -> crate::Result { + headers: &'h [picohttp::Header], + ) -> crate::Result<(HeaderResult, picohttp::Response<'h>)> { let mut response = picohttp::Response { minor_version: 0, status_code, @@ -377,14 +376,7 @@ impl HTTPClient<'_> { headers: picohttp::HeaderList { list: headers }, bytes_read: 0, }; - // SAFETY: see fn doc — erased borrow is deep-copied by `clone_metadata` - // before the backing storage is released. - self.state.pending_response = Some(unsafe { response.detach_lifetime() }); let should_continue = self.handle_response_metadata(&mut response)?; - // handle_response_metadata may mutate `response` (e.g. the 304 rewrite - // for force_last_modified); clone_metadata reads pending_response, so - // re-sync. SAFETY: same lifetime erase as above. - self.state.pending_response = Some(unsafe { response.detach_lifetime() }); // h2/h3 framing delimits the body; chunked transfer-encoding and the // HTTP/1.1 "no Content-Length ⇒ no keep-alive" rule don't apply. self.state.transfer_encoding = Encoding::Identity; @@ -392,11 +384,12 @@ impl HTTPClient<'_> { self.state.response_stage = ResponseStage::Body; } self.state.flags.allow_keepalive = true; - Ok(if should_continue == ShouldContinue::Finished { + let result = if should_continue == ShouldContinue::Finished { HeaderResult::Finished } else { HeaderResult::HasBody - }) + }; + Ok((result, response)) } } @@ -1570,25 +1563,10 @@ impl<'a> HTTPClient<'a> { } /// Common tail of `fail` / `fail_from_h2` / `complete_connecting_process`: /// build the result, reset request state, and dispatch the callback. - /// Factored out so the borrowck reshape (`to_result()` borrows `&mut self` - /// while the post-reset callback wants `&mut self.state` again) lives in - /// one place instead of being open-coded with raw `(*this_ptr).field` at - /// every fail site. fn dispatch_result_and_reset(&mut self, clear_proxy_tunneling: bool) { let callback = self.result_callback; - // reshaped for borrowck — `to_result()`'s `body` field is a - // `&mut MutableString` derived from a NonNull (caller-owned, disjoint - // from `self`'s storage), but its lifetime is tied to `&mut self`. - // Detach so the `state.reset()` reborrow below compiles. - // SAFETY: `body_out_str` points at the caller-owned MutableString that - // outlives this client. NOTE: `state.reset()` below DOES write through - // that same allocation (`(*body_out_str).reset()`, InternalState.rs) - // while `result.body` is a live `&'static mut` to it — this overlap is - // pre-existing (the old open-coded `(*this_ptr).state.reset()` did the - // same); the callback observes the - // post-reset (empty) buffer. Do not read this comment as asserting - // `result.body` and `state.reset()` are disjoint. - let result = unsafe { self.to_result().detach_lifetime() }; + let body = self.state.body_out_str; + let mut 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,6 +1584,9 @@ 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] @@ -3598,55 +3579,17 @@ impl<'a> HTTPClient<'a> { } else { crate::ssl_config::SSLConfig::ZERO }; - // Take ownership of the CONNECT accumulation buffer BEFORE entering - // ProxyTunnel::start. The envelope has been fully consumed by the - // caller (handle_on_data_headers); we leave an empty buffer behind so - // that when the tunnel later re-enters handle_on_data_headers with - // decrypted upstream bytes, the stale CONNECT envelope isn't re-parsed - // as the user-facing response (see #30381). Without this, a split - // CONNECT 200 response (envelope arriving across two TCP reads) stays - // buffered; the tunnel's re-entry appends the decrypted upstream bytes - // onto it, re-parses the envelope as the response (leaking - // proxy-agent / connection: close into response.headers), and hands - // the upstream's raw HTTP/1.1 bytes to the body unparsed. - // - // `start_payload` may alias into this buffer's heap storage on the - // split-read path, but `std::mem::take` swaps only the `Vec` header — - // the heap allocation (and thus the bytes `start_payload` points at) - // stays put until `envelope_buf` is dropped at the end of this - // function. ProxyTunnel::start copies `start_payload` into the TLS BIO - // via start_with_payload -> BIO_write before it returns, so the bytes - // are captured before the drop. - // - // We hold the buffer in a local and drop it AFTER start() rather than - // clearing `self.state.response_message_buffer` afterwards: - // ProxyTunnel::start has synchronous failure paths (SSLWrapper init - // error, or a handshake-traffic error that synchronously fires - // on_close) that call close_and_fail -> fail -> the result callback, - // which can free the AsyncHTTP that embeds `*self`. Touching `self` - // after start() returns would be a use-after-free. - let envelope_buf = std::mem::take(&mut self.state.response_message_buffer); + // The sole caller (`handle_on_data_headers`) has already moved + // `response_message_buffer` into a local, so the CONNECT envelope is + // gone from `self` and `start_payload` borrows that caller local (or + // `incoming_data`), which outlives this call; the #30381 split-envelope + // hazard is handled there. ProxyTunnel::start has synchronous failure + // paths (SSLWrapper init error, or a handshake-traffic error that + // synchronously fires on_close) that call close_and_fail -> fail -> the + // result callback, which can free the AsyncHTTP that embeds `*self`. + debug_assert!(self.state.response_message_buffer.list.capacity() == 0); ProxyTunnel::start::(self, socket, &ssl_options, start_payload); // Must not reference `self` past this point — see comment above. - drop(envelope_buf); - } - - #[inline] - fn handle_short_read( - &mut self, - incoming_data: &[u8], - socket: HttpSocket, - needs_move: bool, - ) { - if needs_move { - let to_copy = incoming_data; - if !to_copy.is_empty() { - // this one will probably be another chunk, so we leave a little extra room - let _ = self.state.response_message_buffer.append(to_copy); // OOM/capacity: fire-and-forget - } - } - - self.set_timeout(&socket); } pub fn handle_on_data_headers( @@ -3660,51 +3603,59 @@ impl<'a> HTTPClient<'a> { "handleOnDataHeader data: {}", BStr::new(incoming_data) ); - // reshaped for borrowck — `to_read` aliases either - // `incoming_data` or `self.state.response_message_buffer`; hold it as a - // `RawSlice` (encapsulated outlives-holder backref, safe `.slice()`) - // so subsequent `&mut self` calls don't trip the checker. - let mut to_read = bun_ptr::RawSlice::new(incoming_data); - macro_rules! to_read { - () => { - to_read.slice() - }; - } - let mut needs_move = true; - if !self.state.response_message_buffer.list.is_empty() { + // Move the accumulation buffer out of `self` so `to_read` can be a + // plain `&[u8]` borrow of either `incoming_data` or the local `buffer`, + // both disjoint from `&mut self`. The short-read paths move it back; + // every other path either drops it (terminal / reset) or lets it drop + // here once `clone_metadata()` has deep-copied the parsed headers. + let mut buffer = std::mem::take(&mut self.state.response_message_buffer); + let needs_move = buffer.list.is_empty(); + let mut to_read: &[u8] = if needs_move { + incoming_data + } else { // this one probably won't be another chunk, so we use appendSliceExact() to avoid over-allocating - let _ = self - .state - .response_message_buffer - .append_slice_exact(incoming_data); - to_read = bun_ptr::RawSlice::new(self.state.response_message_buffer.list.as_slice()); - needs_move = false; + let _ = buffer.append_slice_exact(incoming_data); + buffer.list.as_slice() + }; + + // Persist the unparsed tail for the next `on_data` and re-arm the + // receive timeout. When `needs_move`, `to_read` is a suffix of + // `incoming_data` and is copied into the (currently empty) accumulation + // buffer; otherwise `to_read` is a suffix of `buffer`, so the consumed + // prefix is drained and `buffer` is moved back into state. + macro_rules! short_read { + () => {{ + bun_core::scoped_log!(fetch, "handleShortRead"); + if needs_move { + if !to_read.is_empty() { + // this one will probably be another chunk, so we leave a little extra room + let _ = self.state.response_message_buffer.append(to_read); + } + } else { + let keep = to_read.len(); + buffer + .list + .drain_front(buffer.list.len().saturating_sub(keep)); + self.state.response_message_buffer = buffer; + } + self.set_timeout(&socket); + return; + }}; } - loop { + let shared_resp = scratch::response_headers(); + let mut response = loop { let mut amount_read: usize = 0; - // we reset the pending_response each time wich means that on parse error this will be always be empty - self.state.pending_response = Some(picohttp::Response::default()); - // minimal http/1.1 response is 16 bytes ("HTTP/1.1 200\r\n\r\n") // if less than 16 it will always be a ShortRead - if to_read!().len() < 16 { - bun_core::scoped_log!(fetch, "handleShortRead"); - if !needs_move { - let remaining = to_read!().len(); - let buffer = &mut self.state.response_message_buffer.list; - buffer.drain_front(buffer.len().saturating_sub(remaining)); - to_read = bun_ptr::RawSlice::new(buffer.as_slice()); - } - self.handle_short_read::(to_read!(), socket, needs_move); - return; + if to_read.len() < 16 { + short_read!(); } - let shared_resp = scratch::response_headers(); - let response = match picohttp::Response::parse_parts( - to_read!(), - shared_resp, + let parsed = match picohttp::Response::parse_parts( + to_read, + &mut shared_resp[..], Some(&mut amount_read), ) { Ok(r) => r, @@ -3716,21 +3667,14 @@ impl<'a> HTTPClient<'a> { // `response_message_buffer` growth, so use a generous fixed // cap independent of that knob. const MAX_RESPONSE_HEADER_BUFFER: usize = 1024 * 1024; - if to_read!().len() > MAX_RESPONSE_HEADER_BUFFER { + if to_read.len() > MAX_RESPONSE_HEADER_BUFFER { self.close_and_fail::( crate::Error::ResponseHeadersTooLarge, socket, ); return; } - if !needs_move { - let remaining = to_read!().len(); - let buffer = &mut self.state.response_message_buffer.list; - buffer.drain_front(buffer.len().saturating_sub(remaining)); - to_read = bun_ptr::RawSlice::new(buffer.as_slice()); - } - self.handle_short_read::(to_read!(), socket, needs_move); - return; + short_read!(); } Err(e) => { self.close_and_fail::(e.into(), socket); @@ -3738,20 +3682,11 @@ impl<'a> HTTPClient<'a> { } }; - // we save the successful parsed response - // SAFETY: response borrows SHARED_RESPONSE_HEADERS_BUF / response_message_buffer, - // both of which outlive this fn; widen to 'static for storage. - // Rebind `response` to the detached `'static` copy so it no longer - // borrows `to_read` (lets the `to_read` reassignment below pass - // borrowck — `RawSlice::slice` ties output to `&to_read`). - let response = unsafe { response.detach_lifetime() }; - self.state.pending_response = Some(response); - let bytes_read = - (usize::try_from(response.bytes_read).expect("int cast")).min(to_read.len()); - to_read = bun_ptr::RawSlice::new(&to_read.slice()[bytes_read..]); + (usize::try_from(parsed.bytes_read).expect("int cast")).min(to_read.len()); + to_read = &to_read[bytes_read..]; - if response.status_code == 101 { + if parsed.status_code == 101 { if self.flags.upgrade_state == HTTPUpgradeState::None || (self.flags.proxy_tunneling && self.proxy_tunnel.is_none()) { @@ -3763,18 +3698,17 @@ impl<'a> HTTPClient<'a> { self.flags.upgrade_state = HTTPUpgradeState::Upgraded; // start draining the request body self.flush_stream::(socket); - break; + break parsed; } // handle the case where we have a 100 Continue - if response.status_code >= 100 && response.status_code < 200 { + if parsed.status_code >= 100 && parsed.status_code < 200 { bun_core::scoped_log!(fetch, "information headers"); - self.state.pending_response = None; - if to_read!().is_empty() { + if to_read.is_empty() { if !needs_move { - let buffer = &mut self.state.response_message_buffer.list; - buffer.drain_front(buffer.len()); + buffer.list.clear(); + self.state.response_message_buffer = buffer; } // we only received 1XX responses, we wanna wait for the next status code return; @@ -3783,12 +3717,8 @@ impl<'a> HTTPClient<'a> { continue; } - break; - } - // pending_response is already `Option>` (set just above). - // NOTE: copy (Response is Copy), do NOT .take() — clone_metadata() below - // requires pending_response to remain Some. - let mut response: picohttp::Response<'static> = self.state.pending_response.unwrap(); + break parsed; + }; let should_continue = match self.handle_response_metadata(&mut response) { Ok(s) => s, Err(err) => { @@ -3796,10 +3726,6 @@ impl<'a> HTTPClient<'a> { return; } }; - // handle_response_metadata may mutate `response`; mirror it back so - // clone_metadata() sees the up-to-date headers regardless of the - // content-encoding branch below. - self.state.pending_response = Some(response); if (self.state.content_encoding_i as usize) < response.headers.list.len() && !self.state.flags.did_set_content_encoding @@ -3807,8 +3733,6 @@ impl<'a> HTTPClient<'a> { // if it compressed with this header, it is no longer because we will decompress it self.state.flags.did_set_content_encoding = true; self.state.content_encoding_i = u8::MAX; - // we need to reset the pending response because we removed a header - self.state.pending_response = Some(response); } if should_continue == ShouldContinue::Finished { @@ -3818,7 +3742,7 @@ impl<'a> HTTPClient<'a> { } // this means that the request ended // clone metadata and return the progress at this point - self.clone_metadata(); + self.clone_metadata(&response); // if is chuncked but no body is expected we mark the last chunk self.state.flags.received_last_chunk = true; // if is not we ignore the content_length @@ -3829,14 +3753,14 @@ impl<'a> HTTPClient<'a> { if self.flags.proxy_tunneling && self.proxy_tunnel.is_none() { // we are proxing we dont need to cloneMetadata yet - self.start_proxy_handshake::(socket, to_read!()); + self.start_proxy_handshake::(socket, to_read); return; } // we have body data incoming so we clone metadata and keep going - self.clone_metadata(); + self.clone_metadata(&response); - if to_read!().is_empty() { + if to_read.is_empty() { // no body data yet, but we can report the headers if self.signals.get(signals::Field::HeaderProgress) { self.progress_update::(ctx, socket); @@ -3845,7 +3769,7 @@ impl<'a> HTTPClient<'a> { } if self.state.response_stage == ResponseStage::Body { - let report_progress = match self.handle_response_body(to_read!(), true) { + let report_progress = match self.handle_response_body(to_read, true) { Ok(b) => b, Err(err) => { self.close_and_fail::(err, socket); @@ -3859,7 +3783,7 @@ impl<'a> HTTPClient<'a> { } } else if self.state.response_stage == ResponseStage::BodyChunk { self.set_timeout(&socket); - let report_progress = match self.handle_response_body_chunked_encoding(to_read!()) { + let report_progress = match self.handle_response_body_chunked_encoding(to_read) { Ok(b) => b, Err(err) => { self.close_and_fail::(err, socket); @@ -3904,7 +3828,6 @@ impl<'a> HTTPClient<'a> { // the proxy_tunnel dispatch above: a tunneled target's raw inner-TLS // records must keep reaching the SSLWrapper while parked. if self.state.flags.is_waiting_for_cert_check { - self.state.pending_response = None; self.close_and_fail::(crate::Error::UnexpectedData, socket); return; } @@ -3952,7 +3875,6 @@ impl<'a> HTTPClient<'a> { } ResponseStage::Fail => {} _ => { - self.state.pending_response = None; self.close_and_fail::(crate::Error::UnexpectedData, socket); return; } @@ -4037,50 +3959,36 @@ impl<'a> HTTPClient<'a> { } } - // We have to clone metadata immediately after use - pub fn clone_metadata(&mut self) { - debug_assert!(self.state.pending_response.is_some()); - // `Response<'static>` is `Copy`; bind by value so no borrow - // of `self.state` is held across the `pending_response = None` write - // below. - if let Some(response) = self.state.pending_response { - if let Some(old) = self.state.cloned_metadata.take() { - drop(old); // deinit - } - let mut builder = picohttp::StringBuilder::default(); - response.count(&mut builder); - builder.count(self.url.href); - let _ = builder.allocate(); - // headers_buf is owned by the cloned_response (aka cloned_response.headers) - // `Response::clone` ties its return lifetime to - // `headers: &'a mut [Header]`; leak the box to obtain `'static` so - // the cloned response can be stored in `HTTPResponseMetadata`. - // Reclaimed by `Drop for HTTPResponseMetadata`. - let headers_buf = bun_core::heap::release( - vec![picohttp::Header::ZERO; response.headers.list.len()].into_boxed_slice(), - ); - let cloned_response = response.clone(headers_buf, &mut builder); - - // we clean the temporary response since cloned_metadata is now the owner - self.state.pending_response = None; - - // SAFETY: `href` aliases `builder`'s heap buffer; ownership of that - // buffer is transferred to `owned_buf` immediately below and stored - // alongside `href` in `HTTPResponseMetadata`. - let href = bun_ptr::RawSlice::new(unsafe { builder.append_raw(self.url.href) }); - // Transfer the single backing allocation out of the builder - // (`builder.ptr.?[0..builder.cap]`) so its Drop becomes a no-op. - let owned_buf = builder.move_to_slice(); - self.state.cloned_metadata = Some(HTTPResponseMetadata { - owned_buf, - response: cloned_response, - url: href, - }); - } else { - // we should never clone metadata that dont exists - // we added a empty metadata just in case but will hit the assert - self.state.cloned_metadata = Some(HTTPResponseMetadata::default()); - } + /// Deep-copy `response` (headers, status, and this request's `url.href`) + /// into owned storage on `state.cloned_metadata` so the caller can drop the + /// buffer the parsed slices borrow. + pub fn clone_metadata(&mut self, response: &picohttp::Response<'_>) { + self.state.cloned_metadata = None; + let mut builder = picohttp::StringBuilder::default(); + response.count(&mut builder); + builder.count(self.url.href); + let _ = builder.allocate(); + // headers_buf is owned by the cloned_response (aka cloned_response.headers) + // `Response::clone` ties its return lifetime to + // `headers: &'a mut [Header]`; leak the box to obtain `'static` so + // the cloned response can be stored in `HTTPResponseMetadata`. + // Reclaimed by `Drop for HTTPResponseMetadata`. + let headers_buf = bun_core::heap::release( + vec![picohttp::Header::ZERO; response.headers.list.len()].into_boxed_slice(), + ); + let cloned_response = response.clone(headers_buf, &mut builder); + // SAFETY: `href` aliases `builder`'s heap buffer; ownership of that + // buffer is transferred to `owned_buf` immediately below and stored + // alongside `href` in `HTTPResponseMetadata`. + let href = bun_ptr::RawSlice::new(unsafe { builder.append_raw(self.url.href) }); + // Transfer the single backing allocation out of the builder + // (`builder.ptr.?[0..builder.cap]`) so its Drop becomes a no-op. + let owned_buf = builder.move_to_slice(); + self.state.cloned_metadata = Some(HTTPResponseMetadata { + owned_buf, + response: cloned_response, + url: href, + }); } /// The idle timeout to arm for this request, in seconds (0 = disabled): @@ -4176,14 +4084,6 @@ impl<'a> HTTPClient<'a> { if self.flags.protocol != Protocol::Http1_1 { return self.send_progress_update_multiplexed(); } - // reshaped for borrowck — `to_result()` returns an - // `HTTPClientResult<'_>` whose lifetime is tied to `&mut self` (via the - // `body: &mut MutableString` borrow). Holding that result across the - // `is_done` mutations below would require a second live `&mut Self`, - // which PORTING.md §Forbidden flags as aliased `&mut`. Instead: - // snapshot every owned/Copy field out of the result, drop it, mutate - // `self` directly, then rebuild a fresh `HTTPClientResult` for the - // callback from the snapshotted fields + the restored body. 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 @@ -4191,32 +4091,8 @@ impl<'a> HTTPClient<'a> { let body_snapshot = body_out::take_list(body); let callback = self.result_callback; - let ( - has_more, - redirected, - can_stream, - is_http2, - fail, - dns_error, - dns_hostname, - metadata, - body_size, - certificate_info, - ) = { - let r = self.to_result(); - ( - r.has_more, - r.redirected, - r.can_stream, - r.is_http2, - r.fail, - r.dns_error, - r.dns_hostname, - r.metadata, - r.body_size, - r.certificate_info, - ) - }; // r (and its &mut borrow of self) dropped here + let mut result = self.to_result(); + let has_more = result.has_more; let is_done = !has_more; bun_core::scoped_log!(fetch, "progressUpdate {}", is_done); @@ -4343,23 +4219,8 @@ impl<'a> HTTPClient<'a> { // Restore the body bytes that `state.reset()` cleared. body_out::restore_list(body, body_snapshot); - let async_http = self.parent_async_http(); - // Rebuild the result from snapshotted fields now that all `&mut self` - // mutations are finished — no aliased borrows remain. - let result = HTTPClientResult { - body: body_out::opt_mut(body), - has_more, - redirected, - can_stream, - is_http2, - fail, - dns_error, - dns_hostname, - metadata, - body_size, - certificate_info, - }; - callback.run(async_http, result); + result.body = body_out::opt_mut(body); + callback.run(self.parent_async_http(), result); if has_more { self.maybe_pause_receive(socket); @@ -4381,44 +4242,13 @@ 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); - // reshaped for borrowck — `to_result()` ties `result`'s - // lifetime to `&mut self`, so holding it across the `is_done` mutations - // would require a second live `&mut Self` (aliased UB). Instead snapshot - // every owned/Copy field out of the result, drop it, mutate `self` - // directly, then rebuild a fresh `HTTPClientResult` for the callback. - // See send_progress_update_without_stage_check for the same pattern. 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 ( - has_more, - redirected, - can_stream, - is_http2, - fail, - dns_error, - dns_hostname, - metadata, - body_size, - certificate_info, - ) = { - let r = self.to_result(); - ( - r.has_more, - r.redirected, - r.can_stream, - r.is_http2, - r.fail, - r.dns_error, - r.dns_hostname, - r.metadata, - r.body_size, - r.certificate_info, - ) - }; // r (and its &mut borrow of self) dropped here - let is_done = !has_more; + let mut result = self.to_result(); + let is_done = !result.has_more; bun_core::scoped_log!(fetch, "progressUpdate {}", is_done); if is_done { self.unregister_abort_tracker(); @@ -4430,23 +4260,8 @@ impl<'a> HTTPClient<'a> { } // Restore the body bytes that `state.reset()` cleared. body_out::restore_list(body, body_snapshot); - let async_http = self.parent_async_http(); - // Rebuild the result from snapshotted fields now that all `&mut self` - // mutations are finished — no aliased borrows remain. - let result = HTTPClientResult { - body: body_out::opt_mut(body), - has_more, - redirected, - can_stream, - is_http2, - fail, - dns_error, - dns_hostname, - metadata, - body_size, - certificate_info, - }; - callback.run(async_http, result); + result.body = body_out::opt_mut(body); + callback.run(self.parent_async_http(), result); } /// `do_redirect` minus the per-request socket release/close. The session @@ -4591,7 +4406,14 @@ impl<'a> HTTPClient<'a> { } } - pub fn to_result(&mut self) -> HTTPClientResult<'_> { + /// 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. + pub fn to_result(&mut self) -> HTTPClientResult<'static> { let body_size: BodySize = if self.state.is_chunked_encoding() { BodySize::TotalReceived(self.state.total_body_received) } else if let Some(content_length) = self.state.content_length { @@ -4612,7 +4434,7 @@ impl<'a> HTTPClient<'a> { // transfer ownership of the metadata here return HTTPClientResult { metadata: Some(metadata), - body: body_out::opt_mut(self.state.body_out_str), + body: None, redirected: self.flags.redirected, fail: self.state.fail, dns_error: self.state.dns_error, @@ -4628,7 +4450,7 @@ impl<'a> HTTPClient<'a> { } } HTTPClientResult { - body: body_out::opt_mut(self.state.body_out_str), + body: None, metadata: None, redirected: self.flags.redirected, fail: self.state.fail, @@ -4695,16 +4517,6 @@ impl<'a> HTTPClient<'a> { .get_body_buffer() .append_slice_exact(incoming_data)?; } - - if self.state.response_message_buffer.owns(incoming_data) { - // i'm not sure why this would happen and i haven't seen it happen - // but we should check - debug_assert!( - self.state.get_body_buffer().list.as_ptr() - != self.state.response_message_buffer.list.as_ptr() - ); - self.state.response_message_buffer = MutableString::default(); - } } self.report_progress(incoming_data.len()); @@ -4879,31 +4691,19 @@ impl<'a> HTTPClient<'a> { // using content-encoding per chunk is not supported self.state.chunked_decoder.consume_trailer = 1; - // Capture the length up front so no `&[u8]` aliases the live `&mut [u8]` below. + // `handle_on_data_headers` moves `response_message_buffer` into a + // local before dispatching here, so `incoming_data` never aliases + // `self` and the scratch copy is always sufficient (the dispatcher + // bounds `incoming_data.len()` to the scratch size). let in_len = incoming_data.len(); - let buffer: &mut [u8] = if self.state.response_message_buffer.owns(incoming_data) { - // if we've already copied the buffer once, we can avoid copying it again. - // SAFETY: `incoming_data` is a subslice of `response_message_buffer.list` - // (`owns` just verified). - // `incoming_data.as_ptr() as *mut u8` would carry SharedReadOnly provenance - // (it came from a `&[u8]`) and writing through it is UB. Derive the mutable - // slice from the owning Vec instead so the write has Unique provenance. - let base = self.state.response_message_buffer.list.as_mut_ptr(); - let off = incoming_data.as_ptr() as usize - base as usize; - // SAFETY: `owns()` proved `[base+off, base+off+in_len)` lies within - // `response_message_buffer.list`; `base` carries Unique provenance. - unsafe { bun_core::ffi::slice_mut(base.add(off), in_len) } - } else { - small[0..in_len].copy_from_slice(incoming_data); - &mut small[0..in_len] - }; + let buffer = &mut small[0..in_len]; + buffer.copy_from_slice(incoming_data); let mut bytes_decoded = in_len; // phr_decode_chunked mutates in-place // SAFETY: `buffer` is an exclusive &mut [u8] of len == in_len; offset // len - in_len == 0 is trivially in bounds. `chunked_decoder` is a - // disjoint field of `self.state` (no live borrow of `self` at this - // point — `buffer` is raw-derived or borrows `small`). + // disjoint field of `self.state` (`buffer` borrows `small`). let pret = unsafe { picohttp::phr_decode_chunked( &raw mut self.state.chunked_decoder, diff --git a/src/http_jsc/websocket_client.rs b/src/http_jsc/websocket_client.rs index b5c43240de97..11820e8d61bc 100644 --- a/src/http_jsc/websocket_client.rs +++ b/src/http_jsc/websocket_client.rs @@ -1615,20 +1615,9 @@ impl WebSocket { let ws = Self::new_raw(outgoing, global_this, deflate_params, secure, None); // `adopt_group` takes a closure to write the new socket. - let group = { - // reshaped for borrowck — `rare_data()` borrows `vm` - // mutably and `ws_client_group` also wants a `vm` reference. - let vm_ptr: *mut _ = global_this.bun_vm().as_mut(); - // SAFETY: `rare_data()` returns `&mut RareData` reached through - // `vm.rare_data: Option>`, i.e. a SEPARATE heap - // allocation behind a `Box` — the returned `&mut` does not cover - // any byte of `*vm_ptr` itself, so forming `&*vm_ptr` alongside - // it is non-overlapping under Stacked Borrows. `lazy_group` only - // reads `vm.uws_loop()` / `vm.event_loop_handle` and never touches - // `vm.rare_data`, so the shared `&VirtualMachine` argument cannot - // observe or invalidate the `&mut RareData` receiver. - unsafe { (*vm_ptr).rare_data().ws_client_group::(&*vm_ptr) } - }; + let vm = global_this.bun_vm().as_mut(); + let loop_ = vm.uws_loop(); + let group = vm.rare_data().ws_client_group::(loop_); if !Socket::::adopt_group( tcp, group, diff --git a/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs b/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs index a56ba8b9485f..ebeb277acd26 100644 --- a/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs +++ b/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs @@ -403,13 +403,12 @@ impl HTTPClient { using_proxy ); - // Reshaped for borrowck — `rare_data()` borrows `vm` mutably and - // `ws_upgrade_group` also wants a `vm` reference. See websocket_client.rs. - let group = { - // SAFETY: `rare_data()` returns `&mut RareData` reached through a - // separate Box; the `&*vm_ptr` argument does not overlap. - unsafe { (*vm_ptr).rare_data().ws_upgrade_group::(&*vm_ptr) } - }; + let loop_ = global.bun_vm().uws_loop(); + let group = global + .bun_vm() + .as_mut() + .rare_data() + .ws_upgrade_group::(loop_); let kind: SocketKind = if SSL { SocketKind::WsClientUpgradeTls } else { diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index e2f0840499b3..ed4cbfb877ad 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -6374,21 +6374,10 @@ impl VirtualMachine { self.event_loop_mut().ensure_waker(); - // Note: reshaped for borrowck — `rare_data()` borrows `self` and - // `spawn_ipc_group` then needs `&mut VirtualMachine`. Split via raw - // pointers (disjoint fields) per the existing `Bun__RareData__*` - // accessors in virtual_machine_exports.rs. - #[cfg(not(windows))] - let this: *mut VirtualMachine = self; - #[cfg(not(windows))] let instance: *mut IPCInstance = { - // SAFETY: disjoint borrow — `spawn_ipc_group` only touches the - // embedded `SocketGroup` field + `vm.uws_loop()`. - let group: *mut uws::SocketGroup = unsafe { - let rare = std::ptr::from_mut::((*this).rare_data()); - (*rare).spawn_ipc_group(&*this) - }; + let loop_ = self.uws_loop(); + let group: *mut uws::SocketGroup = self.rare_data().spawn_ipc_group(loop_); // Box the instance first so `data.owner` can name its final // address. diff --git a/src/jsc/rare_data.rs b/src/jsc/rare_data.rs index 02710b20babc..8f215ac18387 100644 --- a/src/jsc/rare_data.rs +++ b/src/jsc/rare_data.rs @@ -743,81 +743,91 @@ impl RareData { } // ── socket groups: lazy init ────────────────────────────────────────── + // + // These take the `uws::Loop` pointer directly (rather than + // `&VirtualMachine`) because every caller reaches `&mut RareData` through + // `vm.rare_data()`, which already holds `&mut VirtualMachine`; requiring a + // second `&VirtualMachine` just to read `vm.uws_loop()` forced a raw-pointer + // split-borrow at every call site. The loop pointer is `Copy` and read + // before `rare_data()` is borrowed, so no aliasing. #[inline] - fn lazy_group<'a>(g: &'a mut SocketGroup, vm: &VirtualMachine) -> &'a mut SocketGroup { + fn lazy_group(g: &mut SocketGroup, loop_: *mut uws::Loop) -> &mut SocketGroup { if g.loop_.is_null() { - g.init(vm.uws_loop(), None, core::ptr::null_mut()); + g.init(loop_, None, core::ptr::null_mut()); } g } - pub fn spawn_ipc_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { - Self::lazy_group(&mut self.spawn_ipc_group, vm) + pub fn spawn_ipc_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { + Self::lazy_group(&mut self.spawn_ipc_group, loop_) } - pub fn test_parallel_ipc_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { - Self::lazy_group(&mut self.test_parallel_ipc_group, vm) + pub fn test_parallel_ipc_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { + Self::lazy_group(&mut self.test_parallel_ipc_group, loop_) } /// One shared group per (VM, ssl) for every `Bun.connect` / `tls.connect` /// client socket. Replaces the old per-connection `us_socket_context_t` /// allocation that was the root of the SSL_CTX-per-connect leak. - pub fn bun_connect_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn bun_connect_group( + &mut self, + loop_: *mut uws::Loop, + ) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.bun_connect_group_tls } else { &mut self.bun_connect_group_tcp }, - vm, + loop_, ) } - pub fn postgres_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn postgres_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.postgres_tls_group } else { &mut self.postgres_group }, - vm, + loop_, ) } - pub fn mysql_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn mysql_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.mysql_tls_group } else { &mut self.mysql_group_ }, - vm, + loop_, ) } - pub fn valkey_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn valkey_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.valkey_tls_group } else { &mut self.valkey_group_ }, - vm, + loop_, ) } - pub fn ws_upgrade_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn ws_upgrade_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.ws_upgrade_tls_group } else { &mut self.ws_upgrade_group_ }, - vm, + loop_, ) } - pub fn ws_client_group(&mut self, vm: &VirtualMachine) -> &mut SocketGroup { + pub fn ws_client_group(&mut self, loop_: *mut uws::Loop) -> &mut SocketGroup { Self::lazy_group( if SSL { &mut self.ws_client_tls_group } else { &mut self.ws_client_group_ }, - vm, + loop_, ) } diff --git a/src/picohttp/lib.rs b/src/picohttp/lib.rs index 4917de2b390f..32bfce35276c 100644 --- a/src/picohttp/lib.rs +++ b/src/picohttp/lib.rs @@ -544,27 +544,6 @@ impl<'a> Default for Response<'a> { } impl<'a> Response<'a> { - /// Widen `status`/`headers` to `'static` for self-referential storage. - /// Field-by-field move (no bitwise reinterpret). - /// - /// # Safety - /// Caller guarantees the response buffer / header storage the slices borrow - /// outlives every read through the returned value. - #[inline] - pub unsafe fn detach_lifetime(self) -> Response<'static> { - Response { - minor_version: self.minor_version, - status_code: self.status_code, - // SAFETY: caller contract. - status: unsafe { &*core::ptr::from_ref::<[u8]>(self.status) }, - headers: HeaderList { - // SAFETY: caller contract. - list: unsafe { &*core::ptr::from_ref::<[Header]>(self.headers.list) }, - }, - bytes_read: self.bytes_read, - } - } - pub fn count(&self, builder: &mut StringBuilder) { builder.count(self.status); @@ -573,18 +552,24 @@ impl<'a> Response<'a> { } } - pub fn clone(&self, headers: &'a mut [Header], builder: &mut StringBuilder) -> Response<'a> { - let mut that = *self; - // SAFETY: see `Header::clone` — caller keeps `builder` alive. - that.status = unsafe { builder.append_raw(self.status) }; - + pub fn clone<'out>( + &self, + headers: &'out mut [Header], + builder: &mut StringBuilder, + ) -> Response<'out> { for (i, header) in self.headers.list.iter().enumerate() { headers[i] = header.clone(builder); } - - that.headers.list = &headers[0..self.headers.list.len()]; - - that + Response { + minor_version: self.minor_version, + status_code: self.status_code, + // SAFETY: see `Header::clone` — caller keeps `builder` alive. + status: unsafe { builder.append_raw(self.status) }, + headers: HeaderList { + list: &headers[0..self.headers.list.len()], + }, + bytes_read: self.bytes_read, + } } pub fn parse_parts( diff --git a/src/runtime/api/bun/js_bun_spawn_bindings.rs b/src/runtime/api/bun/js_bun_spawn_bindings.rs index 6306f9dbf8a5..c255615739ad 100644 --- a/src/runtime/api/bun/js_bun_spawn_bindings.rs +++ b/src/runtime/api/bun/js_bun_spawn_bindings.rs @@ -1520,18 +1520,16 @@ pub(crate) fn spawn_maybe_sync( #[cfg(unix)] if !IS_SYNC { if let Some(mode) = maybe_ipc_mode { - // SAFETY: re-borrow `jsc_vm` through the raw pointer for the nested - // `vm` arg while `rare_data()` holds the outer &mut. - let raw_socket = unsafe { &mut *jsc_vm_ptr } - .rare_data() - .spawn_ipc_group(unsafe { &mut *jsc_vm_ptr }) - .from_fd( - bun_uws::SocketKind::SpawnIpc, - None, - core::mem::size_of::<*mut IPC::SendQueue>() as core::ffi::c_int, - posix_ipc_fd.native(), - true, - ); + // SAFETY: `jsc_vm_ptr` is the live per-thread VM; JS thread. + let vm = unsafe { &mut *jsc_vm_ptr }; + let loop_ = vm.uws_loop(); + let raw_socket = vm.rare_data().spawn_ipc_group(loop_).from_fd( + bun_uws::SocketKind::SpawnIpc, + None, + core::mem::size_of::<*mut IPC::SendQueue>() as core::ffi::c_int, + posix_ipc_fd.native(), + true, + ); if !raw_socket.is_null() { let socket = raw_socket; subprocess.ipc_data.set(Some(IPC::SendQueue::init( diff --git a/src/runtime/cli/test/parallel/Channel.rs b/src/runtime/cli/test/parallel/Channel.rs index 5b9584ad5b41..0d49db566a8d 100644 --- a/src/runtime/cli/test/parallel/Channel.rs +++ b/src/runtime/cli/test/parallel/Channel.rs @@ -114,14 +114,8 @@ impl Channel { /// `SocketKind` value of its own. The per-file isolation swap skips /// `rare.test_parallel_ipc_group` so the coordinator link survives. fn ensure_posix_group(vm: &mut VirtualMachine) -> &mut uws::SocketGroup { - // borrowck split — `rare_data()` mutably borrows `vm`, but - // the group accessor needs `vm` again for `uws_loop()`. The two touch - // disjoint storage (the `Box` payload vs the loop pointer - // field), so a raw-pointer reborrow is sound here. - let rd: *mut bun_jsc::rare_data::RareData = vm.rare_data(); - // SAFETY: `rd` points into `vm`'s boxed RareData, which outlives this - // call; the accessor only reads `vm.uws_loop()` (a separate field). - let g = unsafe { (*rd).test_parallel_ipc_group(vm) }; + let loop_ = vm.uws_loop(); + let g = vm.rare_data().test_parallel_ipc_group(loop_); // First Owner to call wins the vtable; coordinator and worker run in // separate processes so there's never more than one Owner type sharing // this group. diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 812d03f612fd..c4d0238428af 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -547,16 +547,12 @@ impl NewSocket { let this = unsafe { bun_ptr::ThisPtr::new(self.as_ctx_ptr()) }; let _guard = this.ref_guard(); - let vm = self.get_handlers().vm; - // SAFETY: per-thread VM singleton; `VirtualMachine::get()` yields the - // canonical `*mut` (write provenance) — never derive `&mut` from the - // `&'static` borrow stored on Handlers (that's `invalid_reference_casting`). - // No aliasing `&mut` held across the `rare_data()` borrow — `vm` - // reborrowed immutably for the 2nd arg. - let group = VirtualMachine::get() - .as_mut() - .rare_data() - .bun_connect_group::(vm); + // `VirtualMachine::get()` yields the canonical `*mut` (write + // provenance) — never derive `&mut` from the `&'static` borrow stored + // on Handlers (that's `invalid_reference_casting`). + let vm = VirtualMachine::get().as_mut(); + let loop_ = vm.uws_loop(); + let group = vm.rare_data().bun_connect_group::(loop_); let kind: uws::SocketKind = if SSL { uws::SocketKind::BunSocketTls } else { @@ -3493,11 +3489,11 @@ impl NewSocket { // `on_open`/`start_tls_handshake`. let sni: Option<&core::ffi::CStr> = cfg.and_then(|c| c.server_name_cstr()); - // SAFETY: per-thread VM singleton; no aliasing `&mut` held. + let loop_ = vm.uws_loop(); let group = VirtualMachine::get() .as_mut() .rare_data() - .bun_connect_group::(vm); + .bun_connect_group::(loop_); // SAFETY: `raw_socket` is the live `*mut us_socket_t` extracted from // `InternalSocket::Connected` above; `owned_ssl_ctx` is the +1 ref // taken from SecureContext/ssl_ctx_cache and never null here. diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index d71ee5a77bc9..28fbc111edc7 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1514,21 +1514,12 @@ impl JSValkeyClient { let socket_ref = self.ref_scope(); let is_tls = self.client.get().tls != valkey::TLS::None; - // `vm.rare_data()` needs `&mut VirtualMachine`; `client.vm` - // is `&'static`. Cast through raw — the per-thread VM is single-owner - // on the JS thread, and `valkey_group` only touches the embedded - // `SocketGroup` field + `vm.uws_loop()` (disjoint from anything we - // hold). Same pattern as `Bun__RareData__postgresGroup`. - let vm_ptr = std::ptr::from_ref::(self.client.get().vm).cast_mut(); - // SAFETY: per-thread VM, accessed from the JS thread; `rare_data()` - // lazy-inits the box. - let group: *mut uws::SocketGroup = unsafe { - let rare = std::ptr::from_mut::((*vm_ptr).rare_data()); - if is_tls { - (*rare).valkey_group::(&*vm_ptr) - } else { - (*rare).valkey_group::(&*vm_ptr) - } + let vm = self.client.get().vm.as_mut(); + let loop_ = vm.uws_loop(); + let group: *mut uws::SocketGroup = if is_tls { + vm.rare_data().valkey_group::(loop_) + } else { + vm.rare_data().valkey_group::(loop_) }; // Populate `_secure` first, then handle the failure branch outside the @@ -1568,8 +1559,8 @@ impl JSValkeyClient { let ssl_ctx: Option<*mut uws::SslCtx> = match &self.client.get().tls { valkey::TLS::None => None, valkey::TLS::Enabled => { - // SAFETY: `vm_ptr` is the live per-thread VM (see above). - Some(unsafe { crate::jsc_hooks::default_client_ssl_ctx(vm_ptr) }) + // SAFETY: `vm` is the live per-thread VM (see above). + Some(unsafe { crate::jsc_hooks::default_client_ssl_ctx(vm) }) } valkey::TLS::Custom(_) => Some(self._secure.get().unwrap()), }; diff --git a/src/sql_jsc/jsc.rs b/src/sql_jsc/jsc.rs index b7a1938c0d1c..17ea4478be38 100644 --- a/src/sql_jsc/jsc.rs +++ b/src/sql_jsc/jsc.rs @@ -287,9 +287,6 @@ pub(crate) trait VirtualMachineSqlExt { /// bun_io::EventLoopCtx for the JS-thread VM, for KeepAlive::{ref_,unref}. fn vm_ctx(&self) -> bun_io::EventLoopCtx; /// Lazy-init `RareData`'s per-protocol uws [`bun_uws::SocketGroup`]. - /// Encapsulates the `rare_data(&mut self)` / `*_group(.., &VirtualMachine)` - /// borrowck conflict (the two borrows touch field-disjoint state) so the - /// four call sites need no per-site raw-pointer dance. fn postgres_socket_group(&mut self) -> &mut bun_uws::SocketGroup; /// See [`Self::postgres_socket_group`]. fn mysql_socket_group(&mut self) -> &mut bun_uws::SocketGroup; @@ -324,20 +321,13 @@ impl VirtualMachineSqlExt for VirtualMachine { } #[inline] fn postgres_socket_group(&mut self) -> &mut bun_uws::SocketGroup { - // `rare_data()` returns the boxed `&mut RareData` (disjoint allocation); - // `*_group` only reads `vm.uws_loop()`. Route the read-only `vm` - // argument through the JS-thread singleton accessor instead of a - // raw-pointer split-borrow — `VirtualMachine::get()` is `&'static` - // and doesn't borrow `self`, so borrowck is satisfied without a - // per-site raw-pointer deref. - self.rare_data() - .postgres_group::(VirtualMachine::get()) + let loop_ = self.uws_loop(); + self.rare_data().postgres_group::(loop_) } #[inline] fn mysql_socket_group(&mut self) -> &mut bun_uws::SocketGroup { - // See `postgres_socket_group` — singleton `&'static` for the read-only - // `vm` argument avoids the raw-pointer split-borrow. - self.rare_data().mysql_group::(VirtualMachine::get()) + let loop_ = self.uws_loop(); + self.rare_data().mysql_group::(loop_) } } diff --git a/test/js/web/fetch/fetch-proxy-connect-tunnel-split-envelope.test.ts b/test/js/web/fetch/fetch-proxy-connect-tunnel-split-envelope.test.ts index c2df0bc2f585..5be5e450a7ab 100644 --- a/test/js/web/fetch/fetch-proxy-connect-tunnel-split-envelope.test.ts +++ b/test/js/web/fetch/fetch-proxy-connect-tunnel-split-envelope.test.ts @@ -28,8 +28,10 @@ // `Bun.sleep(5)` yield so the kernel flushes the first segment and the // fetch client's HTTP thread consumes it before the second segment lands. -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tls as tlsCert } from "harness"; +import { once } from "node:events"; +import net from "node:net"; // Give the subprocess headroom on slow ASAN CI machines — the combined // setup (Bun.serve TLS + net.createServer + fetch TLS handshake) runs @@ -175,3 +177,79 @@ test("fetch through CONNECT proxy with split 200 envelope surfaces upstream resp }); expect(exitCode).toBe(0); }, 30_000); + +// `handle_on_data_headers` mem::take's the header accumulation buffer into a +// local so `to_read` is a plain borrow. These pin the behaviour of the three +// paths the buffer must survive: the short-read put-back, 1xx interim responses +// consumed from the accumulated buffer, and a chunked body in the same read as +// the end of the header block (now always copied into the 16 KiB scratch rather +// than decoded in place). +describe("handle_on_data_headers split-read header accumulation", () => { + async function serveSplit(splitAt: number | "bytes", wire: string): Promise { + const server = net.createServer(socket => { + socket.setNoDelay(true); + socket.once("data", async () => { + if (splitAt === "bytes") { + for (let i = 0; i < wire.length; i++) { + socket.write(wire[i]); + await new Promise(r => setImmediate(r)); + } + } else { + socket.write(wire.slice(0, splitAt)); + await Bun.sleep(5); + socket.write(wire.slice(splitAt)); + } + socket.end(); + }); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as net.AddressInfo; + try { + return await fetch(`http://127.0.0.1:${port}/`, { keepalive: false }); + } finally { + server.close(); + } + } + + test.concurrent("byte-by-byte headers with a leading 100 Continue", async () => { + const res = await serveSplit( + "bytes", + "HTTP/1.1 100 Continue\r\n\r\n" + "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 5\r\n\r\nhello", + ); + expect({ status: res.status, ct: res.headers.get("content-type"), body: await res.text() }).toEqual({ + status: 200, + ct: "text/plain", + body: "hello", + }); + }); + + test.concurrent("multiple 1xx responses accumulated across reads then final 204", async () => { + const interim = "HTTP/1.1 102 Processing\r\n\r\n"; + // Split lands mid-second-interim so the second read must be appended to the + // already-buffered tail before the loop can consume both and the 204. + const res = await serveSplit( + interim.length + 10, + interim + interim + "HTTP/1.1 204 No Content\r\nX-Foo: bar\r\n\r\n", + ); + expect({ status: res.status, xfoo: res.headers.get("x-foo"), body: await res.text() }).toEqual({ + status: 204, + xfoo: "bar", + body: "", + }); + }); + + test.concurrent("chunked body in the same read as the buffered header tail", async () => { + // Split at 30 so the first read buffers a partial header block and the + // second read brings the rest of the headers plus the whole chunked body in + // one on_data() call, so the body is decoded out of the accumulation buffer. + const res = await serveSplit(30, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n"); + expect({ status: res.status, body: await res.text() }).toEqual({ status: 200, body: "hello" }); + }); + + test.concurrent("content-length body in the same read as the buffered header tail", async () => { + // Non-chunked single-packet body: handle_response_body_from_single_packet + // with the body bytes coming from the accumulation buffer. + const res = await serveSplit(30, "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello"); + expect({ status: res.status, body: await res.text() }).toEqual({ status: 200, body: "hello" }); + }); +});