diff --git a/src/http/AsyncHTTP.rs b/src/http/AsyncHTTP.rs index 31776836f9c5..8d3e886c4e51 100644 --- a/src/http/AsyncHTTP.rs +++ b/src/http/AsyncHTTP.rs @@ -253,6 +253,8 @@ pub struct Options<'a> { pub verbose: Option, pub disable_keepalive: Option, pub disable_decompression: Option, + /// The consumer takes `HTTPClientResult::held_body`. Others get it decoded with the last chunk. + pub takes_held_body: bool, pub max_redirects: Option, pub reject_unauthorized: Option, pub tls_props: Option, @@ -476,6 +478,7 @@ impl<'a> AsyncHTTP<'a> { if let Some(val) = options.disable_decompression { this.client.flags.disable_decompression = val; } + this.client.flags.takes_held_body = options.takes_held_body; if let Some(val) = options.max_redirects { this.client.remaining_redirect_count = (val.min(126) + 1) as i8; } diff --git a/src/http/Decompressor.rs b/src/http/Decompressor.rs index fac59577b7d5..1804ef3c6376 100644 --- a/src/http/Decompressor.rs +++ b/src/http/Decompressor.rs @@ -1,4 +1,3 @@ -use bun_core::MutableString; use bun_http_types::Encoding::Encoding; // The streaming decoders below own only their C-side state and take @@ -59,14 +58,14 @@ impl Decompressor { } /// Feed one body chunk `buffer` through the decoder, appending the - /// decompressed output to `body_out_str` until it holds `max_output` bytes. Creates the + /// decompressed output to `out` until it holds `max_output` bytes. Creates the /// decoder on first call. Returns the input bytes consumed. Returns `ShortRead` when more /// input is needed and the stream is not yet done. pub(crate) fn decompress_chunk( &mut self, encoding: Encoding, buffer: &[u8], - body_out_str: &mut MutableString, + out: &mut Vec, max_output: usize, is_done: bool, ) -> crate::Result { @@ -76,7 +75,6 @@ impl Decompressor { if matches!(self, Decompressor::None) { self.init(encoding, buffer)?; } - let out = &mut body_out_str.list; match self { Decompressor::Zlib(reader) => Ok(reader.decompress(buffer, out, max_output, is_done)?), Decompressor::Brotli(reader) => { diff --git a/src/http/InternalState.rs b/src/http/InternalState.rs index 041c611e109e..67b3b212b8fc 100644 --- a/src/http/InternalState.rs +++ b/src/http/InternalState.rs @@ -19,6 +19,28 @@ fn gzip_trailer_size(buffer: &[u8]) -> Option { .map(|size| u32::from_le_bytes(*size) as usize) } +/// One libdeflate call over a whole gzip body into `size` bytes. False: zlib decodes it. +fn inflate_exact_size( + decompressor: &mut bun_libdeflate_sys::libdeflate::Decompressor, + buffer: &[u8], + size: usize, + out: &mut Vec, +) -> bool { + use bun_libdeflate_sys::libdeflate::{Encoding, Status}; + out.clear(); + // A trailer can lie; the streaming path allocates only what is really there. + if out.try_reserve_exact(size).is_err() { + return false; + } + let result = decompressor.decompress_to_vec(buffer, out, Encoding::Gzip); + // Unconsumed input means a multi-member stream (RFC 1952 §2.2), which libdeflate does not decode. + if result.status == Status::Success && result.read == buffer.len() { + return true; + } + out.clear(); + false +} + // TODO: reduce the size of this struct // Many of these fields can be moved to a packed struct and use less space @@ -250,16 +272,30 @@ impl<'a> InternalState<'a> { /// than failing it: chunked decoder already in the trailers state, or a /// close-delimited response (no Content-Length, no Transfer-Encoding). pub(crate) fn is_body_complete_on_close(&self) -> bool { - // Every byte arrived; only the decode is outstanding. - if self.flags.decompress_output_pending && self.is_done() { - return true; - } if self.is_chunked_encoding() { return bun_picohttp::phr_decode_chunked_is_in_trailers(&self.chunked_decoder) != 0; } self.content_length.is_none() && self.response_stage == HTTPStage::Body } + /// Drops input that nothing will decode. + pub(crate) fn discard_held_input(&mut self) { + self.flags.decompress_output_pending = false; + self.compressed_body.list.clear(); + self.compressed_body_consumed = 0; + } + + /// The rest of a body that is complete on the wire, for `HTTPClientResult::held_body`. + pub(crate) fn take_held_body(&mut self) -> HeldBody { + self.flags.decompress_output_pending = false; + HeldBody { + encoding: self.encoding, + decompressor: core::mem::take(&mut self.decompressor), + input: core::mem::take(&mut self.compressed_body.list), + consumed: core::mem::take(&mut self.compressed_body_consumed), + } + } + /// Mark the body complete and drive `process_body_buffer` one last time /// with `is_final_chunk = true` so a compressed stream that never reached /// stream-end is rejected. Call from every site that flips @@ -324,32 +360,12 @@ impl<'a> InternalState<'a> { break 'libdeflate; } if estimated_size < EXACT_SIZE_INFLATE_MAX { - self.decoded_body.list.clear(); - // A trailer can lie; the streaming path below allocates only what is really there. - if self - .decoded_body - .list - .try_reserve_exact(estimated_size) - .is_err() - { - break 'libdeflate; - } - let result = deflater.decompressor_mut().decompress_to_vec( + still_needs_to_decompress = !inflate_exact_size( + deflater.decompressor_mut(), buffer, + estimated_size, &mut self.decoded_body.list, - bun_libdeflate::Encoding::Gzip, ); - // libdeflate decodes a single gzip member; unconsumed - // input means this is a multi-member stream (RFC 1952 - // §2.2). Let the zlib path handle it. - if result.status == bun_libdeflate::Status::Success - && result.read == buffer.len() - { - still_needs_to_decompress = false; - } else { - self.decoded_body.list.clear(); - } - break 'libdeflate; } } @@ -409,7 +425,7 @@ impl<'a> InternalState<'a> { match self.decompressor.decompress_chunk( self.encoding, buffer, - &mut self.decoded_body, + &mut self.decoded_body.list, max_output, is_done, ) { @@ -492,6 +508,62 @@ impl<'a> InternalState<'a> { } } +/// The undecoded rest of a body whose transport finished, and its decoder. The consumer owns it. +pub struct HeldBody { + encoding: Encoding, + decompressor: Decompressor, + input: Vec, + consumed: usize, +} + +// SAFETY: owns its bytes and its decoder's C state, which no thread has a claim on. +unsafe impl Send for HeldBody {} + +impl HeldBody { + /// Appends decoded bytes to `out` until it holds `max_output`. `Ok(true)`: the body has ended. + pub fn decode(&mut self, out: &mut Vec, max_output: usize) -> Result { + if max_output == usize::MAX && self.inflate_exact_size(out) { + return Ok(true); + } + let input = &self.input[self.consumed..]; + log!("Decompressing {} bytes of a held body\n", input.len()); + self.consumed += + self.decompressor + .decompress_chunk(self.encoding, input, out, max_output, true)?; + let ended = out.len() < max_output + || (self.consumed == self.input.len() && !self.decompressor.is_mid_stream()); + Ok(ended) + } +} + +impl HeldBody { + /// A gzip body that no decoder has started, for a consumer that takes all of it. + fn inflate_exact_size(&mut self, out: &mut Vec) -> bool { + if !bun_core::feature_flags::is_libdeflate_enabled() + || self.encoding != Encoding::Gzip + || !matches!(self.decompressor, Decompressor::None) + || self.consumed != 0 + || !out.is_empty() + { + return false; + } + let Some(size) = gzip_trailer_size(&self.input).filter(|&n| n < EXACT_SIZE_INFLATE_MAX) + else { + return false; + }; + let Some(mut decompressor) = bun_libdeflate_sys::libdeflate::OwnedDecompressor::new() + else { + return false; + }; + log!("Decompressing {} bytes with libdeflate\n", self.input.len()); + if !inflate_exact_size(&mut decompressor, &self.input, size, out) { + return false; + } + self.consumed = self.input.len(); + true + } +} + #[derive(Clone, Copy, PartialEq, Eq)] pub enum HTTPStage { Pending, diff --git a/src/http/ProxyTunnel.rs b/src/http/ProxyTunnel.rs index 34e4e36fb5b2..4fc026feb62f 100644 --- a/src/http/ProxyTunnel.rs +++ b/src/http/ProxyTunnel.rs @@ -506,7 +506,7 @@ fn on_close(ctx: *mut HTTPClient) { && !this.state.flags.is_redirect_pending; let mut fail_err: Option = None; if in_progress && this.state.is_body_complete_on_close() { - match this.state.finalize_body_on_eof() { + match this.finish_body_on_close() { Ok(()) => { // `this` dead (NLL); reborrow via `client_from_ctx` inside. progress_update_for_proxy_socket(ctx, proxy_nn); diff --git a/src/http/Signals.rs b/src/http/Signals.rs index 57a0696ef9e1..104cdacd9c21 100644 --- a/src/http/Signals.rs +++ b/src/http/Signals.rs @@ -21,7 +21,7 @@ pub const BODY_HIGH_WATER_MARK: usize = 256 * 1024; /// moves `Paused -> Flowing` and schedules a resume. The transport applies `Paused` after the /// next read. Two terminal states: `BufferAll` (a consumer wants the whole body) and /// `Abandoned` (nothing will read it; the transport is being shut down, drop what arrives). -/// `Unclaimed`: `Flowing` before a consumer attaches. See `Signals::hold_for_consumer`. +/// `Unclaimed`: `Flowing` before a consumer attaches. A body that completes in it stays undecoded. #[repr(u8)] #[derive(Copy, Clone, PartialEq, Eq, Debug)] pub enum BodyReceiveMode { @@ -88,32 +88,32 @@ impl Signals { .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Paused as u8) } - /// `Flowing`, `Paused` or `Unclaimed`: a consumer takes the body piece by piece. + /// Nothing will read the body, and its consumer is shutting the transport down. #[inline] - pub(crate) fn is_demand_driven(self) -> bool { + pub(crate) fn is_body_abandoned(self) -> bool { self.body_receive_mode .map(bun_ptr::BackRef::from) - .is_some_and(|a| { - matches!( - BodyReceiveMode::from_u8(a.load(Ordering::Acquire)), - BodyReceiveMode::Flowing | BodyReceiveMode::Paused | BodyReceiveMode::Unclaimed - ) - }) + .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Abandoned as u8) + } + + /// No consumer has attached to the body yet. + #[inline] + pub(crate) fn is_body_unclaimed(self) -> bool { + self.body_receive_mode + .map(bun_ptr::BackRef::from) + .is_some_and(|a| a.load(Ordering::Acquire) == BodyReceiveMode::Unclaimed as u8) } - /// `Unclaimed -> Paused`: a whole body waits for the consumer, which resumes the transport. + /// `Flowing`, `Paused` or `Unclaimed`: a consumer takes the body piece by piece. #[inline] - pub(crate) fn hold_for_consumer(self) -> bool { + pub(crate) fn is_demand_driven(self) -> bool { self.body_receive_mode .map(bun_ptr::BackRef::from) .is_some_and(|a| { - a.compare_exchange( - BodyReceiveMode::Unclaimed as u8, - BodyReceiveMode::Paused as u8, - Ordering::AcqRel, - Ordering::Relaxed, + matches!( + BodyReceiveMode::from_u8(a.load(Ordering::Acquire)), + BodyReceiveMode::Flowing | BodyReceiveMode::Paused | BodyReceiveMode::Unclaimed ) - .is_ok() }) } } diff --git a/src/http/lib.rs b/src/http/lib.rs index 78efb36ac771..4fc1f5698c0f 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -59,7 +59,7 @@ pub use http_context::{HTTPContext, HTTPSocket, PeerVerification}; pub use http_request_body::HTTPRequestBody; pub use http_thread::HttpThread as HTTPThread; pub use http_thread::shutdown_for_exit; -pub use internal_state::InternalState; +pub use internal_state::{HeldBody, InternalState}; pub use proxy_tunnel::ProxyTunnel; pub use send_file::SendFile; pub use signals::Signals; @@ -223,6 +223,8 @@ pub struct Flags { pub forced_protocol: Option, pub(crate) h3_retried: bool, pub is_node_http_client: bool, + /// `Options::takes_held_body`. + pub(crate) takes_held_body: bool, } impl Default for Flags { @@ -246,6 +248,7 @@ impl Default for Flags { forced_protocol: None, h3_retried: false, is_node_http_client: false, + takes_held_body: false, } } } @@ -488,6 +491,8 @@ pub struct HTTPClientResult<'a> { /// Boxed: it is large and rare, and every result is moved and dropped /// several times per request. pub proxy_connect_response: Option>, + /// Final result only: what the consumer's budget left undecoded (`Options::takes_held_body`). + pub held_body: Option>, } /// Keep-alive pool partition of the fetch session a request belongs to. @@ -577,6 +582,7 @@ impl<'a> HTTPClientResult<'a> { certificate_info: self.certificate_info, connect_errno: self.connect_errno, proxy_connect_response: self.proxy_connect_response, + held_body: self.held_body, } } } @@ -2149,7 +2155,7 @@ impl<'a> HTTPClient<'a> { return; } if in_progress && self.state.is_body_complete_on_close() { - if let Err(err) = self.state.finalize_body_on_eof() { + if let Err(err) = self.finish_body_on_close() { self.fail(err); return; } @@ -2192,11 +2198,6 @@ impl<'a> HTTPClient<'a> { if self.flags.disable_timeout { return; } - // A fully received body that waits on its consumer expects nothing from the socket. - if self.state.has_pending_compressed() && self.state.is_done() { - socket.set_timeout(0); - return; - } bun_core::scoped_log!(fetch, "Timeout {}\n", BStr::new(self.url.href)); // Terminate (mark dead + close) BEFORE failing, matching // `close_and_fail`: `fail()` dispatches the final result, which frees @@ -4109,33 +4110,58 @@ impl<'a> HTTPClient<'a> { socket.set_timeout(self.effective_idle_timeout_seconds()); } - /// Output budget of one decode pass. h1 only: h2/h3 detach before held input could drain. + /// A compressed body is decoded one budget at a time, as its consumer reads. h1 only. #[inline] - fn decompress_output_cap(&self) -> usize { - if self.flags.protocol == Protocol::Http1_1 && self.signals.is_demand_driven() { - signals::BODY_HIGH_WATER_MARK + fn decodes_on_demand(&self) -> bool { + self.flags.protocol == Protocol::Http1_1 && self.signals.is_demand_driven() + } + + /// Output budget of one decode pass. The last pass has `HELD_BODY_MIN` more, or no budget without a `HeldBody` taker. + #[inline] + fn decompress_output_cap(&self, is_final_chunk: bool) -> usize { + if !self.decodes_on_demand() { + return usize::MAX; + } + if !is_final_chunk { + return signals::BODY_HIGH_WATER_MARK; + } + if self.flags.takes_held_body { + signals::BODY_HIGH_WATER_MARK + Self::HELD_BODY_MIN } else { usize::MAX } } + /// A rest that decodes to less than this ends on the HTTP thread: it is not worth a work-pool job. + const HELD_BODY_MIN: usize = 64 * 1024; + /// Decodes what has arrived under the consumer's budget. Returns whether to report bytes. fn process_received_body(&mut self, is_final_chunk: bool) -> crate::Result { - let mut max_output = self.decompress_output_cap(); - if max_output != usize::MAX && self.state.encoding.is_compressed() { - // Nothing is decoded for a paused consumer (a tunnelled socket keeps reading anyway). - if self.signals.is_receive_paused() { - self.state.flags.decompress_output_pending = true; + let mut max_output = self.decompress_output_cap(is_final_chunk); + if self.state.encoding.is_compressed() { + // Nothing will read it, and its transport is being shut down. + if self.signals.is_body_abandoned() { + self.state.discard_held_input(); return Ok(false); } - // A body that one libdeflate call can inflate waits whole for its consumer. - if is_final_chunk && self.state.wants_exact_size_inflate() { - if self.signals.hold_for_consumer() { - self.state.flags.decompress_output_pending = true; - return Ok(false); + if max_output != usize::MAX { + let paused = self.signals.is_receive_paused(); + if is_final_chunk && self.state.wants_exact_size_inflate() { + // A body that one libdeflate call can inflate goes whole to its consumer. + if paused || self.signals.is_body_unclaimed() { + self.state.flags.decompress_output_pending = true; + return Ok(false); + } + // A consumer attached after the cap was read. + max_output = self.decompress_output_cap(is_final_chunk); + } else if paused { + // Nothing is decoded for a paused consumer (a tunnelled socket keeps reading anyway). + if !is_final_chunk { + self.state.flags.decompress_output_pending = true; + return Ok(false); + } + max_output = Self::HELD_BODY_MIN; } - // A consumer attached after the cap was read. - max_output = self.decompress_output_cap(); } } // `process_body_buffer` takes `&mut self.state`, so the bytes move out first. @@ -4177,6 +4203,12 @@ impl<'a> HTTPClient<'a> { self.set_timeout(&socket); } + /// The transport ended, and with it a close-delimited body. + pub(crate) fn finish_body_on_close(&mut self) -> crate::Result<()> { + self.state.flags.received_last_chunk = true; + self.process_received_body(true).map(drop) + } + pub(crate) fn drain_response_body(&mut self, socket: HttpSocket) { if self.pump_held_body::(socket) { let ctx = self.get_ssl_ctx::(); @@ -4205,18 +4237,15 @@ impl<'a> HTTPClient<'a> { } // A consumer that paused again gets another resume when it unpauses. - let pumped = self.state.has_pending_compressed() && !self.signals.is_receive_paused(); - if pumped { - let is_final = self.state.is_done(); - if let Err(err) = self.process_received_body(is_final) { + if self.state.has_pending_compressed() && !self.signals.is_receive_paused() { + // Not the final chunk: a body that is complete on the wire has ended the request. + if let Err(err) = self.process_received_body(false) { self.close_and_fail::(err, socket); return false; } } - // A pump that ends the body has to say so even with no bytes (a stream trailer alone). - let ended = pumped && self.state.is_done() && !self.state.has_pending_compressed(); - !self.state.decoded_body.list.is_empty() || ended + !self.state.decoded_body.list.is_empty() } fn send_progress_update_without_stage_check( @@ -4566,6 +4595,14 @@ impl<'a> HTTPClient<'a> { None }; let certificate_info = self.state.certificate_info.take(); + // check if we are reporting cert errors, do not have a fail state and we are not done + let has_more = + certificate_info.is_some() || (self.state.fail.is_none() && !self.state.is_done()); + // The transport has all of the body. What is not decoded yet is the consumer's to decode. + let held_body = + (!has_more && self.state.fail.is_none() && self.state.has_pending_compressed()) + .then(|| Box::new(self.state.take_held_body())); + debug_assert!(held_body.is_none() || self.flags.takes_held_body); if certificate_info.is_none() { if let Some(metadata) = self.state.cloned_metadata.take() { // transfer ownership of the metadata here @@ -4579,8 +4616,8 @@ impl<'a> HTTPClient<'a> { dns_hostname: self.state.dns_hostname.take(), connect_errno: self.state.connect_errno, proxy_connect_response: None, - has_more: self.state.fail.is_none() - && (!self.state.is_done() || self.state.has_pending_compressed()), + has_more, + held_body, body_size, certificate_info: None, can_stream: (self.state.request_stage == RequestStage::Body @@ -4600,10 +4637,8 @@ impl<'a> HTTPClient<'a> { dns_hostname: self.state.dns_hostname.take(), connect_errno: self.state.connect_errno, proxy_connect_response, - // check if we are reporting cert errors, do not have a fail state and we are not done - has_more: certificate_info.is_some() - || (self.state.fail.is_none() - && (!self.state.is_done() || self.state.has_pending_compressed())), + has_more, + held_body, body_size, certificate_info, // we can stream the request_body at this stage @@ -4631,7 +4666,7 @@ impl<'a> HTTPClient<'a> { && let Some(len) = content_length && incoming_data.len() >= len // The single-packet path decodes the whole body with no output budget. - && !(self.state.encoding.is_compressed() && self.signals.is_demand_driven()) + && !(self.state.encoding.is_compressed() && self.decodes_on_demand()) { self.handle_response_body_from_single_packet(&incoming_data[0..len])?; Ok(true) @@ -4723,7 +4758,6 @@ impl<'a> HTTPClient<'a> { // Close-delimited bodies still need per-packet decompression, but // a non-streaming consumer must not see per-packet progress: the // terminal callback (on close) is the first to carry metadata. - let is_done = is_done && !self.state.has_pending_compressed(); return Ok(is_done || (processed && is_streaming)); } Ok(false) @@ -4736,7 +4770,7 @@ impl<'a> HTTPClient<'a> { let small_len = 16 * 1024usize; if incoming_data.len() <= small_len && self.state.get_body_buffer().list.is_empty() - && !(self.state.encoding.is_compressed() && self.signals.is_demand_driven()) + && !(self.state.encoding.is_compressed() && self.decodes_on_demand()) { self.handle_response_body_chunked_encoding_from_single_packet(incoming_data) } else { @@ -4808,12 +4842,11 @@ impl<'a> HTTPClient<'a> { // Done _ => { self.state.flags.received_last_chunk = true; - let processed = self.process_received_body(true)?; + self.process_received_body(true)?; self.report_progress(buffer_len); - // A held body ends when `drain_response_body` has pumped it dry, not here. - return Ok(processed || !self.state.has_pending_compressed()); + return Ok(true); } } } diff --git a/src/runtime/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index b02934a1dd14..6584cec46029 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -137,6 +137,12 @@ pub struct FetchTasklet { /// buffer used to stream response to JS pub(crate) scheduled_response_buffer: MutableString, + /// `HTTPClientResult::held_body`, from when `on_progress_update` has it. JS thread only. + held_body: JsCell, + /// What `heap::into_raw` returned for this tasklet, for work that outlives a `&self` caller. + allocation: *mut FetchTasklet, + /// A held body arrived: the HTTP thread is done with this fetch, the response body is not. + pub(crate) is_transport_done: bool, /// response weak ref we need this to track the response JS lifetime pub(crate) response: jsc::Weak, /// native response ref if we still need it when JS is discarted @@ -929,6 +935,15 @@ impl FetchTasklet { bun_output::scoped_log!(FetchTasklet, "onProgressUpdate"); self.mutex.lock(); self.has_schedule_callback.store(false, Ordering::Relaxed); + if let Some(held_body) = self.result.held_body.take() { + self.held_body.set(HeldBodyState::AtRest(held_body)); + self.is_transport_done = true; + } + if matches!(self.held_body.get(), HeldBodyState::Unread) { + self.held_body.set(HeldBodyState::None); + self.result.has_more = false; + self.result.fail = Some(http::Error::Aborted); + } let is_done = !self.result.has_more; let vm = self.global_this.bun_vm(); @@ -945,7 +960,8 @@ impl FetchTasklet { } } self.mutex.unlock(); - if is_done { + // Nothing will read a held body either, and it is all that is left of this fetch. + if is_done || self.drop_held_body_at_rest() { // SAFETY: `self` is the live heap tasklet; we hold a ref. FetchTasklet::deref(std::ptr::from_mut(self)); } @@ -967,6 +983,10 @@ impl FetchTasklet { .with_mut(|poll_ref| poll_ref.unref(bun_io::js_vm_ctx())); // SAFETY: `this` is the live heap tasklet; we hold a ref. FetchTasklet::deref(std::ptr::from_mut(this)); + } else if this.is_transport_done { + // As above: nothing takes the rest of an upload. The response body goes on. + this.cancel_request_body_sink(JSValue::UNDEFINED); + this.drive_held_body(); } }; @@ -975,7 +995,7 @@ impl FetchTasklet { if let Err(err) = self.start_request_stream() { // The VM is being stopped: leave like the `!script_allowed()` gate above does. self.mutex.unlock(); - if is_done { + if is_done || self.drop_held_body_at_rest() { // SAFETY: `self` is the live heap tasklet; we hold a ref. FetchTasklet::deref(std::ptr::from_mut(self)); } @@ -1823,6 +1843,8 @@ impl FetchTasklet { .with_mut(|poll_ref| poll_ref.ref_(bun_io::js_vm_ctx())); if this.signal_store.receive_all() { this.schedule_receive_resume(); + } else { + this.drive_held_body(); } } @@ -1830,6 +1852,102 @@ impl FetchTasklet { if let Some(http_) = self.http.as_ref() { http::http_thread().schedule_receive_resume(http_.async_http_id); } + self.drive_held_body(); + } + + /// Produces a held body. Called wherever the HTTP thread would hear from the consumer. + fn drive_held_body(&self) { + if !matches!(self.held_body.get(), HeldBodyState::AtRest(_)) { + return; + } + let mode = self.signal_store.body_receive_mode(); + if self.is_body_unread(mode) { + return self.end_unread_held_body(); + } + if matches!(mode, BodyReceiveMode::Paused | BodyReceiveMode::Unclaimed) { + return; + } + let HeldBodyState::AtRest(held) = self.held_body.replace(HeldBodyState::InPass) else { + return; + }; + let cx = self + .global_this + .js_thread(self.global_this.bun_vm().context_of(self.context)); + self.ref_(); + jsc::Job::::schedule( + &cx, + HeldBodyPass { + held, + max_output: if mode == BodyReceiveMode::BufferAll { + usize::MAX + } else { + BODY_HIGH_WATER_MARK + }, + out: Vec::new(), + ended: Ok(false), + }, + HeldBodyPassOwner(self.allocation), + ); + } + + /// Nothing will read the rest of the body: its fetch was aborted, or its response is gone. + fn is_body_unread(&self, mode: BodyReceiveMode) -> bool { + mode == BodyReceiveMode::Abandoned || self.signal_store.aborted.load(Ordering::Relaxed) + } + + /// Queues the end of the body. This task runs in a stopped context too: the body gets settled. + fn end_unread_held_body(&self) { + self.held_body.set(HeldBodyState::Unread); + let task = Task::init(self.allocation); + // SAFETY: JS thread; the loop is this VM's. + unsafe { (*self.global_this.bun_vm().event_loop()).enqueue_task(task) }; + } + + fn drop_held_body_at_rest(&self) -> bool { + let at_rest = matches!(self.held_body.get(), HeldBodyState::AtRest(_)); + if at_rest { + self.held_body.set(HeldBodyState::None); + } + at_rest + } + + /// A `HeldBodyPass` is back: what `callback` and the task it posts do for the HTTP thread. + fn on_held_body_pass(&mut self, pass: HeldBodyPass) -> JsResult<()> { + let unread = self.is_body_unread(self.signal_store.body_receive_mode()); + self.mutex.lock(); + match pass.ended { + Ok(ended) if !unread => { + self.result.has_more = !ended; + self.held_body.set(if ended { + HeldBodyState::None + } else { + HeldBodyState::AtRest(pass.held) + }); + let scheduled = &mut self.scheduled_response_buffer; + if scheduled.list.is_empty() { + scheduled.list = pass.out; + } else if scheduled.list.len() < pass.out.len() { + let mut out = pass.out; + out.splice(0..0, scheduled.list.drain(..)); + scheduled.list = out; + } else { + bun_core::handle_oom(scheduled.write(&pass.out)); + } + if !ended && scheduled.list.len() >= BODY_HIGH_WATER_MARK { + self.signal_store.pause_receive(); + } + } + ended => { + self.held_body.set(HeldBodyState::None); + self.result.has_more = false; + self.result.fail = Some(match ended { + Err(err) if !unread => err, + _ => http::Error::Aborted, + }); + } + } + self.mutex.unlock(); + self.on_progress_update() } fn to_body_value(&mut self) -> BodyValue { @@ -1990,6 +2108,9 @@ impl FetchTasklet { request_body: fetch_options.body, request_body_streaming_buffer: None, scheduled_response_buffer: MutableString::default(), + held_body: JsCell::new(HeldBodyState::None), + allocation: core::ptr::null_mut(), + is_transport_done: false, response: jsc::Weak::default(), native_response: JsCell::new(None), response_stream: Default::default(), @@ -2082,6 +2203,7 @@ impl FetchTasklet { let fetch_tasklet_ptr = bun_core::heap::into_raw(fetch_tasklet); // SAFETY: just allocated; exclusive access until returned let fetch_tasklet = unsafe { &mut *fetch_tasklet_ptr }; + fetch_tasklet.allocation = fetch_tasklet_ptr; // This task gets queued on the HTTP thread. // `AsyncHTTP::init` takes several `&'static [u8]` borrows @@ -2141,6 +2263,7 @@ impl FetchTasklet { idle_timeout_seconds: fetch_options.idle_timeout_seconds, disable_keepalive: Some(fetch_options.disable_keepalive), disable_decompression: Some(fetch_options.disable_decompression), + takes_held_body: true, max_redirects: fetch_options.max_redirects, reject_unauthorized: Some(fetch_options.reject_unauthorized), verbose: Some(fetch_options.verbose), @@ -2292,7 +2415,7 @@ impl FetchTasklet { data: RequestBodyChunk<'_>, high_water_mark: usize, ) -> Writable { - if self.signal_aborted() || self.request_body_mismatched { + if self.signal_aborted() || self.is_transport_done || self.request_body_mismatched { return Writable::Done; } // An empty chunk is a no-op on every framing path. It must not reach @@ -2376,8 +2499,10 @@ impl FetchTasklet { FetchTasklet::deref(this_ptr); return; } + // An upload that ends after its response is complete on the wire changes nothing. + let is_over = self.is_transport_done || self.signal_store.aborted.load(Ordering::Relaxed); if let Some(js_error) = err { - if self.signal_store.aborted.load(Ordering::Relaxed) || self.abort_reason.has() { + if is_over || self.abort_reason.has() { // SAFETY: `this_ptr` derived from live `&mut self`; we hold a ref. FetchTasklet::deref(this_ptr); return; @@ -2387,7 +2512,7 @@ impl FetchTasklet { } self.abort_task(); } else { - if self.signal_store.aborted.load(Ordering::Relaxed) { + if is_over { // SAFETY: `this_ptr` derived from live `&mut self`; we hold a ref. FetchTasklet::deref(this_ptr); return; @@ -2439,6 +2564,7 @@ impl FetchTasklet { if let Some(http_) = self.http.as_deref() { http::http_thread().schedule_shutdown(http_); } + self.drive_held_body(); true } @@ -2466,7 +2592,10 @@ impl FetchTasklet { let global_this = self.global_this; self.abort_reason.set(&global_this, reason); } - self.abort_task(); + // An abort would take the rest of the response body with it. + if !self.is_transport_done { + self.abort_task(); + } if let Some(sink) = self.sink_mut() { sink.pending.result = Writable::Done; sink.pending.run(); @@ -2477,7 +2606,7 @@ impl FetchTasklet { } if is_native { // No pump promise exists to balance the `+1` from - // `start_request_stream`; `aborted` is set above so + // `start_request_stream`; the fetch is over (see `write_end_request`), so // `write_end_request(Some(_))` is just the balancing deref. self.write_end_request(Some(reason)); } @@ -2565,6 +2694,16 @@ impl FetchTasklet { // SAFETY: lifetime erasure for non-body fields; `body` is stored as // `&'static []` so no borrow escapes. task_ref.result = unsafe { result.detach_lifetime() }; + // Read once: the JS thread can abandon the body at any point of this callback. + let abandoned = task_ref.signal_store.body_receive_mode() == BodyReceiveMode::Abandoned; + // The transport is done, the body is not: `on_progress_update` takes the rest from here. + if task_ref.result.held_body.is_some() { + if abandoned { + task_ref.result.held_body = None; + } else { + task_ref.result.has_more = true; + } + } // 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. task_ref.result.can_stream = task_ref.result.can_stream || prev_can_stream; @@ -2590,7 +2729,7 @@ impl FetchTasklet { let success = task_ref.result.is_success(); - if task_ref.signal_store.body_receive_mode() == BodyReceiveMode::Abandoned { + if abandoned { if task_ref.scheduled_response_buffer.list.capacity() > 0 { task_ref.scheduled_response_buffer = MutableString::default(); } @@ -2762,6 +2901,58 @@ impl FetchTasklet { } } +/// Where the rest of a body is once its transport is done. +enum HeldBodyState { + /// The HTTP thread produces the body, or the body has ended. + None, + AtRest(Box), + /// A `HeldBodyPass` has it. + InPass, + /// Nothing will read it, and the `on_progress_update` that ends the body is queued. + Unread, +} + +/// One decode pass over a held body, on the work pool. +struct HeldBodyPass { + held: Box, + max_output: usize, + out: Vec, + ended: Result, +} + +/// The tasklet a `HeldBodyPass` reports to, and a ref on it. +struct HeldBodyPassOwner(*mut FetchTasklet); + +// SAFETY: a ref on the tasklet, which is the JS thread's; used and dropped there. +unsafe impl bun_jsc::job::JsAffine for HeldBodyPassOwner {} + +impl Drop for HeldBodyPassOwner { + /// Released unrun: the context that fetched stopped, or the VM is going. + fn drop(&mut self) { + // SAFETY: the pass holds a ref; JS thread. + unsafe { (*self.0).end_unread_held_body() }; + FetchTasklet::deref(self.0); + } +} + +impl jsc::JobContext for HeldBodyPass { + type OffThread = Self; + type Js = HeldBodyPassOwner; + + fn run(pass: &mut Self, done: jsc::Completion) -> Option> { + pass.ended = pass.held.decode(&mut pass.out, pass.max_output); + Some(done) + } + + fn then(pass: Self, owner: HeldBodyPassOwner, _: &jsc::JsThread<'_>) -> JsResult<()> { + let tasklet = core::mem::ManuallyDrop::new(owner).0; + // SAFETY: the pass holds a ref; JS thread. + let result = unsafe { (*tasklet).on_held_body_pass(pass) }; + FetchTasklet::deref(tasklet); + result + } +} + pub(crate) struct FetchOptions { pub method: Method, pub(crate) headers: Headers, diff --git a/test/js/web/fetch/fetch-backpressure.test.ts b/test/js/web/fetch/fetch-backpressure.test.ts index 7ca9d0400944..476fa9da9c3c 100644 --- a/test/js/web/fetch/fetch-backpressure.test.ts +++ b/test/js/web/fetch/fetch-backpressure.test.ts @@ -559,14 +559,26 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not }; } - // Close-delimited, and the origin never closes: for the client this body does not end. - async function serveBomb(enc: Enc, secure: boolean) { + // How the origin ends the body. "never": close-delimited and the origin never closes, so for + // the client this body does not end. The others send the whole body and close at once, so the + // end of the transport reaches a client that still holds nearly all of the body undecoded. + type Ending = "never" | "content-length" | "chunked" | "close-delimited"; + async function serveBomb(enc: Enc, secure: boolean, ending: Ending = "never") { const bomb = bombFor(enc); + const framing = + ending === "content-length" + ? `Content-Length: ${bomb.length}\r\n` + : ending === "chunked" + ? "Transfer-Encoding: chunked\r\n" + : ""; const handler = (s: import("node:net").Socket) => { s.on("error", () => {}); s.once("data", () => { - s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\nConnection: close\r\n\r\n`); + s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\n${framing}Connection: close\r\n\r\n`); + if (ending === "chunked") s.write(`${bomb.length.toString(16)}\r\n`); s.write(bomb); + if (ending === "chunked") s.write("\r\n0\r\n\r\n"); + if (ending !== "never") s.end(); }); }; const server = await listening(secure ? createTlsServer(tls, handler) : createTcpServer(handler)); @@ -605,8 +617,10 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not const base = process.memoryUsage.rss(); let peak = 0; const sample = () => (peak = Math.max(peak, process.memoryUsage.rss() - base)); - const res = await fetch(url, opts); - const reader = res.body.getReader(); + const body = opts.s3 + ? new Bun.S3Client({ accessKeyId: "t", secretAccessKey: "t", endpoint: url, bucket: "b" }).file("k").stream() + : (await fetch(url, opts)).body; + const reader = body.getReader(); const first = await reader.read(); for (let last = sample(), stable = 0; stable < 3; ) { await Bun.sleep(20); @@ -625,10 +639,17 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not process.stdout.write(JSON.stringify({ got, peak, zeros })); `; + // An ambient NO_PROXY that lists 127.0.0.1 makes fetch() ignore its `proxy` option. + const clientEnv = { ...bunEnv }; + for (const key of ["NO_PROXY", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"]) { + delete clientEnv[key]; + delete clientEnv[key.toLowerCase()]; + } + async function runClient(url: string, opts: object, script: string) { await using proc = Bun.spawn({ cmd: [bunExe(), "-e", `const url=${JSON.stringify(url)};const opts=${JSON.stringify(opts)};${script}`], - env: bunEnv, + env: clientEnv, stdout: "pipe", stderr: "pipe", }); @@ -664,11 +685,156 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not }, ); - test("zstd through a CONNECT proxy: a reader that takes a little holds a little", async () => { - await using server = await serveBomb("zstd", true); - await using proxy = await serveConnectProxy(); - const opts = { proxy: proxy.url, tls: { rejectUnauthorized: false } }; - expectBounded(await runClient(server.url, opts, READ_A_LITTLE)); + // The S3 client reads a bucket's compressed object through the same decoder. + test("gzip from S3: a reader that takes a little holds a little", async () => { + await using server = await serveBomb("gzip", false); + expectBounded(await runClient(server.url, { s3: true }, READ_A_LITTLE)); + }); + + // bun does not pause a tunnelled socket, so all of the body and then the origin's close reach + // the client at once. The end of the transport must not decode what the reader has not asked + // for: the request ends there, and the Response keeps the rest of the body undecoded. + test.each(["never", "content-length", "chunked", "close-delimited"] as Ending[])( + "zstd through a CONNECT proxy, body ending %s: a reader that takes a little holds a little", + async ending => { + await using server = await serveBomb("zstd", true, ending); + await using proxy = await serveConnectProxy(); + const opts = { proxy: proxy.url, tls: { rejectUnauthorized: false } }; + expectBounded(await runClient(server.url, opts, READ_A_LITTLE)); + expect(proxy.connects()).toBe(1); + }, + ); + + // The origin sends the whole of a body at once and keeps the connection: the request ends with + // most of the body undecoded, and the Response keeps that part. + async function serveWhole(body: Buffer) { + const head = `HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${body.length}\r\n\r\n`; + const server = await listening( + createTcpServer(s => { + s.on("error", () => {}); + // Answers each request head. What else arrives is an upload it does not wait for. + s.on("data", data => { + if (data.includes(" HTTP/1.1\r\n")) s.write(Buffer.concat([Buffer.from(head), body])); + }); + }), + ); + return { ...server, url: `http://127.0.0.1:${server.port}/` }; + } + + // The reader is eight budgets into the body when it reaches the damage, long after the + // request ended. The error has to reach it there: no hang, and no body that just stops. + test("gzip: damage in the part the Response kept undecoded rejects the read", async () => { + const member = gzipSync(Buffer.alloc(8 * MARK), { level: 9 }); + const damaged = Buffer.from(member); + damaged.fill(0xff, 20, 60); + await using server = await serveWhole(Buffer.concat([member, damaged])); + const script = /* js */ ` + const res = await fetch(url, opts); + let got = 0, code; + try { + for await (const chunk of res.body) got += chunk.byteLength; + } catch (e) { + code = e.code; + } + process.stdout.write(JSON.stringify({ got, code })); + `; + const { got, code, exitCode } = await runClient(server.url, {}, script); + expect({ got, code }).toEqual({ got: 8 * MARK, code: "ZlibError" }); + expect(exitCode).toBe(0); + }); + + test("an abort reaches the part of a body that the Response kept undecoded", async () => { + await using server = await serveWhole(gzipSync(Buffer.alloc(64 * MARK))); + + const reading = new AbortController(); + const reader = (await fetch(server.url, { signal: reading.signal })).body!.getReader(); + const first = await reader.read(); + reading.abort(); + const rest = (async () => { + while (!(await reader.read()).done); + return "ended"; + })().catch(e => e.name); + + const untouched = new AbortController(); + const res = await fetch(server.url, { signal: untouched.signal }); + untouched.abort(); + + expect({ + first: first.value!.byteLength > 0, + rest: await rest, + untouched: await res.text().catch(e => e.name), + }).toEqual({ first: true, rest: "AbortError", untouched: "AbortError" }); + }); + + // The request is over when its response is complete on the wire, so an upload that is still + // underway ends there, as it does for a body that is not compressed. The Response is not over. + test("an upload still underway ends with the request, not with the part the Response kept undecoded", async () => { + await using server = await serveWhole(gzipSync(Buffer.alloc(16 * MARK, 65))); + const cancelled = Promise.withResolvers(); + const upload = new ReadableStream({ + pull: controller => controller.enqueue(new Uint8Array(1024)), + cancel: () => cancelled.resolve(), + }); + const res = await fetch(server.url, { method: "POST", body: upload, duplex: "half" } as RequestInit); + await cancelled.promise; + const bytes = await res.bytes(); + expect({ length: bytes.byteLength, allA: isAllA(bytes) }).toEqual({ length: 16 * MARK, allA: true }); + }); + + // A host may keep a Response that a Bun.ModuleGraph's code fetched. dispose() aborts the fetch, + // and the body the host still holds has to end with it before what produced it is freed. + test("a Response outlives the Bun.ModuleGraph that fetched it: the part it kept undecoded ends", async () => { + await using server = await serveWhole(gzipSync(Buffer.alloc(64 * MARK))); + using dir = tempDir("held-body-graph", { + "app.ts": `export const get = (url: string) => fetch(url);`, + "host.ts": ` + const graph = new Bun.ModuleGraph({}); + const app = await graph.import(import.meta.dir + "/app.ts"); + const res: Response = await graph.run(() => app.get(process.argv[2])); + graph.dispose(); + // dispose() queued the task that ends the body. It runs before the loop gets here. + for (let i = 0; i < 3; i++) await new Promise(resolve => setImmediate(resolve)); + process.stdout.write(await res.text().catch(e => e.name)); + `, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "host.ts", server.url], + env: clientEnv, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("AbortError"); + expect(exitCode).toBe(0); + }); + + // One Response keeps its part at rest; the other has a reader, so a decode pass is out on + // another thread when the worker goes. + test("a worker can go away while its Responses keep parts of bodies undecoded", async () => { + await using server = await serveWhole(gzipSync(Buffer.alloc(256 * MARK))); + const worker = /* js */ ` + const { parentPort, workerData } = require("node:worker_threads"); + (async () => { + globalThis.atRest = await fetch(workerData.url); + const reader = (await fetch(workerData.url)).body.getReader(); + await reader.read(); + parentPort.postMessage("ready"); + while (!(await reader.read()).done); + })(); + `; + const script = /* js */ ` + const { Worker } = require("node:worker_threads"); + const worker = new Worker(${JSON.stringify(worker)}, { eval: true, workerData: { url } }); + const ready = await new Promise(resolve => worker.once("message", resolve)); + await worker.terminate(); + process.stdout.write(JSON.stringify({ ready })); + `; + const { ready, stderr, exitCode } = await runClient(server.url, {}, script); + expect(stderr).toBe(""); + expect(ready).toBe("ready"); + expect(exitCode).toBe(0); }); // A live stream: two flushed messages in one packet, then the origin goes quiet with the frame @@ -785,28 +951,27 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not // Content-Length bodies end with the origin's FIN, which reaches a client that still holds // most of the body undecoded. Chunked ones stay open. - async function serveBody(kind: Kind, chunked: boolean) { + async function serveBody(kind: Kind, chunked: boolean, secure = false) { const body = bodyFor(kind); const enc = kind === "br-hq" ? "br" : kind; - const server = await listening( - createTcpServer(s => { - s.on("error", () => {}); - s.once("data", () => { - if (chunked) { - s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\nTransfer-Encoding: chunked\r\n\r\n`); - s.write(`${body.length.toString(16)}\r\n`); - s.write(body); - s.write("\r\n0\r\n\r\n"); - } else { - s.write( - `HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n`, - ); - s.end(body); - } - }); - }), - ); - return { ...server, url: `http://127.0.0.1:${server.port}/` }; + const handler = (s: import("node:net").Socket) => { + s.on("error", () => {}); + s.once("data", () => { + if (chunked) { + s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\nTransfer-Encoding: chunked\r\n\r\n`); + s.write(`${body.length.toString(16)}\r\n`); + s.write(body); + s.write("\r\n0\r\n\r\n"); + } else { + s.write( + `HTTP/1.1 200 OK\r\nContent-Encoding: ${enc}\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n`, + ); + s.end(body); + } + }); + }; + const server = await listening(secure ? createTlsServer(tls, handler) : createTcpServer(handler)); + return { ...server, url: `${secure ? "https" : "http"}://127.0.0.1:${server.port}/` }; } const STREAM = /* js */ ` @@ -852,13 +1017,67 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not expect(exitCode).toBe(0); }); }); + + // Through a tunnel the whole body and the origin's close are in before the reader's second + // pull, so nearly all of it comes from the part that the Response kept undecoded. + test.each([ + ["a streaming reader", STREAM, { several: true }], + ["res.bytes()", BUFFER, {}], + ])("zstd through a CONNECT proxy, origin closes after the body: %s", async (_, script, seen) => { + await using server = await serveBody("zstd", false, true); + await using proxy = await serveConnectProxy(); + const opts = { proxy: proxy.url, tls: { rejectUnauthorized: false } }; + const { stderr, exitCode, ...result } = await runClient(server.url, opts, script); + expect(stderr).toBe(""); + expect(result).toEqual({ total: SIZE, digest, ...seen }); + expect(proxy.connects()).toBe(1); + expect(exitCode).toBe(0); + }); + }); + + // A pass over a held body is a work-pool job, and it logs "... of a held body". A rest that + // is short is not worth one: the HTTP thread decodes it with the last of the body. + test.skipIf(!isDebug).each([ + ["a short rest ends on the HTTP thread", MARK + 32 * 1024, false], + ["a long rest goes to its consumer", 4 * MARK, true], + ])("%s", async (_, size, held) => { + const body = brotliCompressSync(Buffer.alloc(size, "alpha beta gamma delta lorem ipsum ")); + const reply = Buffer.concat([ + Buffer.from(`HTTP/1.1 200 OK\r\nContent-Encoding: br\r\nContent-Length: ${body.length}\r\n\r\n`), + body, + ]); + const srv = createTcpServer(socket => { + socket.on("error", () => {}); + socket.on("data", () => socket.write(reply)); + }); + await using server = await listening(srv); + const script = /* js */ ` + const res = await fetch(${JSON.stringify(`http://127.0.0.1:${server.port}/`)}); + let total = 0; + for await (const chunk of res.body) total += chunk.byteLength; + console.log("RESULT", total); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...clientEnv, BUN_DEBUG_HTTPInternalState: "1" }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // Output.scoped writes to whichever stream it chose at init; scan both. + const output = stdout + stderr; + expect({ + result: output.match(/RESULT .*/g), + held: /Decompressing \d+ bytes of a held body/.test(output), + }).toEqual({ result: [`RESULT ${size}`], held }); + expect(exitCode).toBe(0); }); // A gzip body that arrives whole, and whose trailer says it inflates to between 512 KB and // 32 MB, is worth one exact-size libdeflate call instead of zlib passes. A reader's budget rules // that call out, and such a body is usually here before anyone has said how the Response is - // read. So the client holds it undecoded until a consumer attaches: `.bytes()` gets the one - // call, a reader gets its passes. + // read. So the request ends and the Response keeps the body undecoded until a consumer + // attaches: `.bytes()` gets the one call, a reader gets its passes. describe("a gzip body that arrived whole waits for its consumer", () => { const SIZE = 1024 * 1024 + 77; const raw = Buffer.alloc(SIZE, "alpha beta gamma delta lorem ipsum "); @@ -970,7 +1189,7 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not expect(await consume(res)).toBe(digest); }); - // A tunnelled socket is never paused. The consumer's resume reaches the client all the same. + // The same through a tunnel: the Response has the body, so the consumer needs no socket. test.each([ ["res.bytes()", `[await (await fetch(url, opts)).bytes()]`], ["a streaming reader", `await Array.fromAsync((await fetch(url, opts)).body)`], @@ -1006,8 +1225,7 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not expect(exitCode).toBe(0); }); - // The socket of a held body is paused, so the client sees this close only once a consumer - // resumes it. + // The request of a held body has ended, so this close is of an idle connection. test.each(consumers)("%s takes a held body after the origin closed the connection", async (_, consume) => { await using server = await serveWhole("content-length", "with the head"); const res = await fetch(server.url); @@ -1015,9 +1233,7 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not expect(await consume(res)).toBe(digest); }); - // A tunnelled socket is never paused, so the close reaches the client while it still holds - // the body. `server.closed` settles once the client has closed its side, which is after it - // handled the end of the body. Only then does the consumer attach. + // `server.closed` settles once the client has closed its side. Only then does the consumer attach. test.each([ ["res.bytes()", `[await res.bytes()]`], ["a streaming reader", `await Array.fromAsync(res.body)`], @@ -1083,15 +1299,70 @@ describe.concurrent("fetch() receive backpressure — the decompressor does not expect({ stdout, stderr, exitCode }).toEqual({ stdout: "", stderr: "", exitCode: 0 }); }); - test("held, and its Response is collected: the fetch is aborted", async () => { - await using server = await serveWhole("content-length", "with the head"); - // Its own frame, so that nothing on this one still refers to the response afterwards. - async function abandon() { - expect((await fetch(server.url)).status).toBe(200); - } - await abandon(); - await collectUntil(server.closed); - }); + // The request has ended, so there is no fetch to abort. Only a debug build has the log. + test.skipIf(!isDebug)( + "held, and its Response is collected: nothing is decoded, and the fetch is freed", + async () => { + const N = 8; + const reply = Buffer.concat([ + Buffer.from(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${wire.length}\r\n\r\n`), + wire, + ]); + const srv = createTcpServer(socket => { + socket.on("error", () => {}); + socket.on("data", () => socket.write(reply)); + }); + await using server = await listening(srv); + const script = /* js */ ` + const url = ${JSON.stringify(`http://127.0.0.1:${server.port}/`)}; + // Its own frame, so that nothing on the caller's still refers to the response afterwards. + async function abandon() { + if ((await fetch(url)).status !== 200) throw new Error("status"); + } + for (let i = 0; i < ${N}; i++) await abandon(); + let done = false; + (async () => { + for await (const line of console) if (line === "done") done = true; + })(); + // Bounded, so that a tasklet that is never freed fails the count below. + for (let turn = 0; turn < 200 && !done; turn++) { + Bun.gc(true); + await new Promise(resolve => setImmediate(resolve)); + } + // A body that is read logs its pass, so the scan below is known to see one. + await (await fetch(url)).bytes(); + process.exit(0); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...clientEnv, BUN_DEBUG_FetchTasklet: "1", BUN_DEBUG_HTTPInternalState: "1" }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + let freed = 0; + let passesWhileHeld = -1; + const passes: string[] = []; + // Output.scoped writes to whichever stream it chose at init; scan both. + const scan = async (stream: ReadableStream) => { + for await (const line of forEachLine(stream)) { + if (/Decompressing \d+ bytes/.test(line)) passes.push(line); + if (/\[FetchTasklet\] deinit/i.test(line) && ++freed === N) { + passesWhileHeld = passes.length; + proc.stdin.write("done\n"); + proc.stdin.end(); + } + } + }; + await Promise.all([scan(proc.stdout), scan(proc.stderr)]); + expect({ freed: Math.min(freed, N), passesWhileHeld, passesOnceRead: passes.length }).toEqual({ + freed: N, + passesWhileHeld: 0, + passesOnceRead: 1, + }); + expect(await proc.exited).toBe(0); + }, + ); }); }); @@ -1919,6 +2190,46 @@ describe.serial("fetch() receive backpressure — an unread body hands its conne } }); + // A compressed body is decoded as it is read, so all of it can be in while most of it is still + // undecoded. That ends the request: the Response keeps the rest, the connection goes back. + test("a compressed body that is complete on the wire, Response still held: its connection is reused", async () => { + const SIZE = 4 * MARK; + const body = gzipSync(Buffer.alloc(SIZE, 65)); + let connections = 0; + const sockets = new Set(); + const srv = createTcpServer(socket => { + connections++; + sockets.add(socket); + socket.on("error", () => {}); + socket.on("data", () => { + socket.write( + Buffer.concat([ + Buffer.from(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${body.length}\r\n\r\n`), + body, + ]), + ); + }); + }); + srv.listen(0, "127.0.0.1"); + await once(srv, "listening"); + try { + const url = `http://127.0.0.1:${(srv.address() as import("node:net").AddressInfo).port}/`; + const N = 8; + const responses: Response[] = []; + for (let i = 0; i < N; i++) responses.push(await fetch(url)); + // Not exactly 1: a request can start before the previous body's last packet was taken. + // A body that pins its connection until it is read needs N of them. + expect(connections).toBeLessThan(N / 2); + for (const res of responses) { + const bytes = await res.bytes(); + expect({ length: bytes.byteLength, allA: isAllA(bytes) }).toEqual({ length: SIZE, allA: true }); + } + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise(r => srv.close(() => r(undefined))); + } + }); + // Not close-delimited: such a body ends with its connection, so there is nothing to reuse. for (const framing of ["content-length", "chunked"] as Framing[]) { test(`a short ${framing} body, Response still held: it is received, and its connection is reused`, async () => {