diff --git a/Cargo.lock b/Cargo.lock index f3d43b32701..e24d69a60dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -924,7 +924,7 @@ dependencies = [ [[package]] name = "curl" version = "0.4.49" -source = "git+https://github.com/lexmount/curl-rust?rev=a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758#a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758" +source = "git+https://github.com/lexmount/curl-rust?rev=dc8c170e4370772c0085c08a67dda0fa86efca70#dc8c170e4370772c0085c08a67dda0fa86efca70" dependencies = [ "curl-sys", "libc", @@ -938,7 +938,7 @@ dependencies = [ [[package]] name = "curl-sys" version = "0.4.87+curl-8.19.0" -source = "git+https://github.com/lexmount/curl-rust?rev=a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758#a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758" +source = "git+https://github.com/lexmount/curl-rust?rev=dc8c170e4370772c0085c08a67dda0fa86efca70#dc8c170e4370772c0085c08a67dda0fa86efca70" dependencies = [ "brotlic-sys", "cc", diff --git a/Cargo.toml b/Cargo.toml index abdb35ba20c..d870d4d9d9a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -104,8 +104,8 @@ fontique = { git = "https://github.com/lexmount/parley", rev = "a18cfc3ba6638d44 subsetter = { git = "https://github.com/lexmount/subsetter", rev = "8d3815245835af309559c3b4a872d6a0a54913a9" } url = { git = "https://github.com/lexmount/rust-url", rev = "ca9fcad2b99f20c30c28a618ad78410f6fd31161" } xml5ever = { path = "vendor/xml5ever-0.39.0" } -curl = { git = "https://github.com/lexmount/curl-rust", rev = "a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758" } -curl-sys = { git = "https://github.com/lexmount/curl-rust", rev = "a0ea59f5ca1b1e4bc8627a28c29463ac96aeb758" } +curl = { git = "https://github.com/lexmount/curl-rust", rev = "dc8c170e4370772c0085c08a67dda0fa86efca70" } +curl-sys = { git = "https://github.com/lexmount/curl-rust", rev = "dc8c170e4370772c0085c08a67dda0fa86efca70" } cookie = { git = "https://github.com/ldm0/cookie-rs", branch = "priority" } v8 = { path = "vendor/v8-152.2.0" } deno_v8 = { path = "vendor/deno_v8-0.3.0" } diff --git a/moli-fetch/src/blocking/collectors.rs b/moli-fetch/src/blocking/collectors.rs index fcecd394969..12e4a62ce50 100644 --- a/moli-fetch/src/blocking/collectors.rs +++ b/moli-fetch/src/blocking/collectors.rs @@ -1,4 +1,4 @@ -use std::{str, time::Duration}; +use std::{num::NonZeroUsize, str, time::Duration}; use anyhow::{Error, Result}; use curl::easy::{Easy2, Handler, WriteError}; @@ -186,6 +186,7 @@ pub struct StreamingResponseCollector { max_response_size: Option, response_too_large: bool, response_bytes_received: usize, + declared_identity_body_length: Option, cache_body_writer: Option, cache_plan: Option, started: bool, @@ -257,7 +258,7 @@ pub struct RawStreamingResponseCollector { max_response_size: Option, response_too_large: bool, response_bytes_received: usize, - declared_identity_body_length: Option, + declared_identity_body_length: Option, cache_body_writer: Option, cache_plan: Option, started: bool, @@ -301,6 +302,7 @@ impl StreamingResponseCollector { max_response_size: None, response_too_large: false, response_bytes_received: 0, + declared_identity_body_length: None, cache_body_writer: None, cache_plan: None, started: false, @@ -342,6 +344,7 @@ impl StreamingResponseCollector { self.max_response_size = max_response_size; self.response_too_large = false; self.response_bytes_received = 0; + self.declared_identity_body_length = None; self.cache_body_writer = cache_body_writer; self.cache_plan = None; self.started = false; @@ -369,6 +372,7 @@ impl StreamingResponseCollector { redirect_chain: Vec, network_request_extra_info: Option, cache_plan: Option, + response_progress: ResponseProgress, ) { self.begin_request( max_response_size, @@ -379,6 +383,7 @@ impl StreamingResponseCollector { redirect_chain, None, ); + self.response_progress = response_progress; self.network_request_extra_info = network_request_extra_info; self.cache_plan = cache_plan; } @@ -501,10 +506,12 @@ impl StreamingResponseCollector { self.client_hint_restart_requested = true; return false; } + let declared_body_length = identity_encoded_content_length(&self.headers); + self.declared_identity_body_length = declared_body_length.and_then(NonZeroUsize::new); // Create the cache writer only after final response headers prove the // response is cacheable. self.maybe_create_cache_body_writer(); - self.maybe_emit_start(); + self.maybe_emit_start(declared_body_length == Some(0)); if response_body_ends_at_headers(self.status) && self.started { self.finish_streaming_body(); self.header_terminated = true; @@ -532,7 +539,7 @@ impl StreamingResponseCollector { } } - fn maybe_emit_start(&mut self) { + fn maybe_emit_start(&mut self, body_is_empty: bool) { if self.started || self.response_too_large { return; } @@ -546,6 +553,9 @@ impl StreamingResponseCollector { { return; } + if body_is_empty || response_body_ends_at_headers(self.status) { + self.response_progress.commit(); + } self.started = true; if let Some(start_tx) = self.start_tx.take() { let _ = start_tx.send(Ok(StreamingHtmlResponseStart { @@ -899,8 +909,9 @@ impl RawStreamingResponseCollector { self.client_hint_restart_requested = true; return false; } - self.declared_identity_body_length = identity_encoded_content_length(&self.headers); - if self.declared_identity_body_length == Some(0) { + let declared_body_length = identity_encoded_content_length(&self.headers); + self.declared_identity_body_length = declared_body_length.and_then(NonZeroUsize::new); + if declared_body_length == Some(0) { self.response_progress.commit(); } self.maybe_create_cache_body_writer(); @@ -1067,7 +1078,7 @@ impl Handler for ResponseCollector { self.status_text = Some(parsed.status_text); self.body.clear(); if let Some(progress) = &self.response_progress { - if parsed.status < 200 { + if is_interim_response_status(parsed.status) { progress.mark_interim_response_started(); } else { progress.mark_response_started(); @@ -1114,6 +1125,13 @@ impl Handler for StreamingResponseCollector { } self.response_bytes_received += data.len(); + if self.started + && self + .declared_identity_body_length + .is_some_and(|length| length.get() == self.response_bytes_received) + { + self.response_progress.commit(); + } self.write_cache_body_bytes(data); self.stream_body_bytes(data); if self.cancel_handle.is_cancelled() { @@ -1137,9 +1155,10 @@ impl Handler for StreamingResponseCollector { self.status_text = Some(parsed.status_text); self.negotiated_http_version = parsed.version; self.response_bytes_received = 0; + self.declared_identity_body_length = None; self.response_too_large = false; self.cookie_set_reports.clear(); - if self.status < 200 { + if is_interim_response_status(self.status) { self.response_progress.mark_interim_response_started(); } else { self.response_progress.mark_response_started(); @@ -1189,7 +1208,10 @@ impl Handler for RawStreamingResponseCollector { } self.response_bytes_received += data.len(); - if self.declared_identity_body_length == Some(self.response_bytes_received) { + if self + .declared_identity_body_length + .is_some_and(|length| length.get() == self.response_bytes_received) + { self.response_progress.commit(); } self.write_cache_body_bytes(data); @@ -1217,10 +1239,10 @@ impl Handler for RawStreamingResponseCollector { self.response_bytes_received = 0; self.response_too_large = false; self.cookie_set_reports.clear(); - if self.status >= 200 { - self.response_progress.mark_response_started(); - } else { + if is_interim_response_status(self.status) { self.response_progress.mark_interim_response_started(); + } else { + self.response_progress.mark_response_started(); } return true; } @@ -1343,6 +1365,49 @@ mod tests { assert!(!easy.get_mut().progress(0.0, 0.0, 0.0, 0.0)); } + #[test] + fn html_zero_length_commits_only_for_an_accepted_response() { + for (status, headers, committed) in [ + ("HTTP/1.1 200 OK", vec!["Content-Length: 0"], true), + ("HTTP/1.1 200 OK", vec![], false), + ( + "HTTP/1.1 302 Found", + vec!["Location: /next", "Content-Length: 0"], + false, + ), + ] { + let cancel = FetchCancelHandle::new(); + let (start_tx, _start_rx) = oneshot::channel(); + let (body_tx, _body_rx) = mpsc::unbounded_channel(); + let mut collector = StreamingResponseCollector::new( + new_shared_browser_cookie_store(), + start_tx, + body_tx, + cancel.clone(), + ); + collector.begin_request( + None, + Url::parse("https://example.test/body").unwrap(), + NetworkCookieRequestContext::top_level_navigation("GET"), + None, + true, + Vec::new(), + None, + ); + assert!(collector.header(format!("{status}\r\n").as_bytes())); + for header in headers { + assert!(collector.header(format!("{header}\r\n").as_bytes())); + } + assert!(collector.header(b"\r\n")); + assert_eq!(cancel.response_completion_is_committed(), committed); + assert_eq!( + cancel.response_timing().response_end_at().is_some(), + committed + ); + assert_eq!(cancel.cancel_if_response_pending(), !committed); + } + } + #[test] fn raw_streaming_exact_identity_content_length_commits_response_completion() { let cancel_handle = FetchCancelHandle::new(); diff --git a/moli-fetch/src/cancellation.rs b/moli-fetch/src/cancellation.rs index 4cf4838c794..bd454db69be 100644 --- a/moli-fetch/src/cancellation.rs +++ b/moli-fetch/src/cancellation.rs @@ -17,27 +17,34 @@ const RESPONSE_CANCELLED: u8 = 2; /// never redirects an existing writer to a different transfer. #[derive(Debug, Clone, Default)] pub(crate) struct ResponseProgress { - state: Arc, - timing: Arc>, + inner: Arc, +} + +#[derive(Debug, Default)] +struct ResponseProgressState { + state: AtomicU8, + timing: Mutex, } impl ResponseProgress { pub(crate) fn mark_request_started(&self) { - self.timing + self.inner + .timing .lock() .request_started_at .get_or_insert_with(Instant::now); } pub(crate) fn mark_interim_response_started(&self) { - self.timing + self.inner + .timing .lock() .first_interim_response_started_at .get_or_insert_with(Instant::now); } pub(crate) fn mark_response_body_received(&self) { - self.timing.lock().body_bytes_received = true; + self.inner.timing.lock().body_bytes_received = true; } pub(crate) fn commit(&self) { @@ -46,7 +53,7 @@ impl ResponseProgress { } fn commit_state(&self) { - let _ = self.state.compare_exchange( + let _ = self.inner.state.compare_exchange( RESPONSE_PENDING, RESPONSE_COMMITTED, Ordering::AcqRel, @@ -55,14 +62,16 @@ impl ResponseProgress { } pub(crate) fn mark_response_started(&self) { - self.timing + self.inner + .timing .lock() .response_started_at .get_or_insert_with(Instant::now); } fn finish_timing(&self) { - self.timing + self.inner + .timing .lock() .response_end_at .get_or_insert_with(Instant::now); @@ -75,11 +84,11 @@ impl ResponseProgress { } fn is_committed(&self) -> bool { - self.state.load(Ordering::Acquire) == RESPONSE_COMMITTED + self.inner.state.load(Ordering::Acquire) == RESPONSE_COMMITTED } fn cancel_if_pending(&self) -> bool { - self.state.compare_exchange( + self.inner.state.compare_exchange( RESPONSE_PENDING, RESPONSE_CANCELLED, Ordering::AcqRel, @@ -102,6 +111,8 @@ impl FetchResponseTiming { pub fn body_bytes_received(self) -> bool { self.body_bytes_received } + /// The transport's pre-request boundary, after connection/TLS setup and + /// before sending the request. Pure cache hits do not produce this boundary. pub fn request_started_at(self) -> Option { self.request_started_at } @@ -226,7 +237,7 @@ impl FetchCancelHandle { } pub fn response_timing(&self) -> FetchResponseTiming { - *self.response_progress().timing.lock() + *self.response_progress().inner.timing.lock() } #[cfg(test)] diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index d13cab77aa1..86456b33b36 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -1012,6 +1012,7 @@ impl RuntimeOwner { anyhow::Error, ), > { + job.response_progress = job.cancel_handle.begin_response_progress(); if let Err(error) = job.request.validate_request_mode_for_url(&job.current_url) { return Err((Box::new(job), None, error)); } @@ -1138,6 +1139,7 @@ impl RuntimeOwner { job.request.redirect_chain.clone(), request_extra_info.clone(), cache_plan, + job.response_progress.clone(), ); collector.set_client_hint_response_policy(prepared_request.response_policy); @@ -1926,6 +1928,7 @@ impl RuntimeOwner { return; } + job.response_progress.commit(); let cache_body_writer = { let streaming = easy .get_mut() @@ -2717,6 +2720,7 @@ struct ActiveRawStreamingTransferContext { pub(crate) struct FetchTransferHandler { response: FetchResponseCollector, + response_body_observed: bool, network_observation_recorder: Option, proxy_connect_response_recorder: ProxyConnectResponseRecorder, } @@ -2728,6 +2732,15 @@ enum FetchResponseCollector { } impl FetchTransferHandler { + fn mark_response_body_received(&mut self) { + if !self.response_body_observed + && let Some(timing) = self.response_progress() + { + timing.mark_response_body_received(); + self.response_body_observed = true; + } + } + fn response_progress(&self) -> Option<&crate::cancellation::ResponseProgress> { match &self.response { FetchResponseCollector::Buffered(collector) => collector.response_progress(), @@ -2751,6 +2764,7 @@ impl FetchTransferHandler { fn new(response: FetchResponseCollector) -> Self { Self { response, + response_body_observed: false, network_observation_recorder: None, proxy_connect_response_recorder: ProxyConnectResponseRecorder::default(), } @@ -2803,6 +2817,7 @@ impl FetchTransferHandler { network_observation_recorder: Option, capture_proxy_connect_response: bool, ) { + self.response_body_observed = false; self.network_observation_recorder = network_observation_recorder; self.proxy_connect_response_recorder .begin_transfer(capture_proxy_connect_response); @@ -2825,7 +2840,23 @@ impl FetchTransferHandler { } impl Handler for FetchTransferHandler { + fn prereq( + &mut self, + _primary_ip: &str, + _local_ip: &str, + _primary_port: u16, + _local_port: u16, + ) -> bool { + if let Some(timing) = self.response_progress() { + timing.mark_request_started(); + } + true + } + fn write(&mut self, data: &[u8]) -> std::result::Result { + if !data.is_empty() { + self.mark_response_body_received(); + } match &mut self.response { FetchResponseCollector::Buffered(collector) => collector.write(data), FetchResponseCollector::Streaming(collector) => collector.write(data), @@ -2845,10 +2876,8 @@ impl Handler for FetchTransferHandler { } fn progress(&mut self, dltotal: f64, dlnow: f64, ultotal: f64, ulnow: f64) -> bool { - if dlnow > 0.0 - && let Some(timing) = self.response_progress() - { - timing.mark_response_body_received(); + if dlnow > 0.0 { + self.mark_response_body_received(); } match &mut self.response { FetchResponseCollector::Buffered(collector) => { @@ -2869,9 +2898,6 @@ impl Handler for FetchTransferHandler { let is_proxy_connect = self .proxy_connect_response_recorder .record_outgoing_header_block(data); - if !is_proxy_connect && let Some(timing) = self.response_progress() { - timing.mark_request_started(); - } if !is_proxy_connect && let Some(recorder) = self.network_observation_recorder.as_ref() { @@ -2893,14 +2919,14 @@ fn configure_network_observation( capture_proxy_connect_response: bool, ) -> Result<()> { let recorder = request.network_observation_recorder().cloned(); - let verbose = recorder.is_some() - || capture_proxy_connect_response - || easy.get_ref().response_progress().is_some(); + let verbose = recorder.is_some() || capture_proxy_connect_response; if let Some(recorder) = recorder.as_ref() { recorder.set_current_request_cookie_report(request_cookie_report.cloned()); } easy.get_mut() .begin_transfer(recorder, capture_proxy_connect_response); + easy.prereq(true) + .context("failed to configure curl pre-request callback")?; easy.verbose(verbose) .context("failed to configure curl network observation") } @@ -2962,10 +2988,19 @@ fn complete_streaming_html_job(job: StreamingRuntimeJob, response: Response) { } fn complete_cached_streaming_html_job( - job: StreamingRuntimeJob, + mut job: StreamingRuntimeJob, cached: CachedStreamingResponseLookup, request_cookie_report: Option, ) { + // Redirects and retries keep the response channels in the reused easy. + // Return them to the job before publishing a final cache hit. + if let Some(mut easy) = job.easy.take() { + (job.started_tx, job.body_tx) = easy + .get_mut() + .streaming_mut() + .expect("cached HTML response should use streaming collector") + .take_response_channels(); + } let redirected = !job.request.redirect_chain.is_empty(); let redirect_chain = job.request.redirect_chain.clone(); let CachedStreamingResponseLookup { @@ -2979,6 +3014,7 @@ fn complete_cached_streaming_html_job( let final_url = match Url::parse(&final_url) { Ok(final_url) => final_url, Err(error) => { + job.response_progress.commit(); if let Some(started_tx) = job.started_tx { let _ = started_tx.send(Err(anyhow!( "failed to parse cached response final url: {error}" @@ -2991,6 +3027,7 @@ fn complete_cached_streaming_html_job( } }; + let response_timing = job.response_progress.mark_cached_response_available(); if let Some(started_tx) = job.started_tx { let _ = started_tx.send(Ok(StreamingHtmlResponseStart { status_text: metadata.status_text, @@ -3012,6 +3049,7 @@ fn complete_cached_streaming_html_job( } else { Ok(()) }; + response_timing.finish(); let _ = job.completion_tx.send(completion); } @@ -3686,16 +3724,18 @@ fn fail_streaming_job(job: StreamingRuntimeJob, error: anyhow::Error) { } fn fail_streaming_job_with_easy( - job: StreamingRuntimeJob, - mut easy: Option>, + mut job: StreamingRuntimeJob, + easy: Option>, error: anyhow::Error, ) { + let mut easy = easy.or_else(|| job.easy.take()); let error = network_fetch_failure_for_request( &job.request, &job.current_url, &job.request.redirect_chain, error, ); + job.response_progress.commit(); if let Some(easy) = easy.as_mut() && let Some(streaming) = easy.get_mut().streaming_mut() { @@ -3790,6 +3830,12 @@ fn request_panics_for_testing(request: &Request) -> bool { }) } +#[cfg(test)] +mod response_timing_tests; + +#[cfg(test)] +mod request_timing_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/moli-fetch/src/runtime/request_timing_tests.rs b/moli-fetch/src/runtime/request_timing_tests.rs new file mode 100644 index 00000000000..6532fde702f --- /dev/null +++ b/moli-fetch/src/runtime/request_timing_tests.rs @@ -0,0 +1,181 @@ +use super::*; +use std::{ + ffi::{c_char, c_int, c_void}, + io::Write, +}; + +extern "C" fn count_debug_calls( + _easy: *mut curl_sys::CURL, + _kind: curl_sys::curl_infotype, + _data: *mut c_char, + _len: usize, + counter: *mut c_void, +) -> c_int { + // SAFETY: tests retain the boxed counter until after Easy2 is dropped, and + // only read it after perform has returned (or its thread has been joined). + unsafe { *counter.cast::() += 1 }; + 0 +} + +fn install_debug_counter(easy: &mut Easy2, counter: &mut usize) { + let callback: curl_sys::curl_debug_callback = count_debug_calls; + // SAFETY: callback has curl_debug_callback's ABI; each caller retains the + // counter allocation for the full lifetime of this easy handle. + unsafe { + assert_eq!( + curl_sys::curl_easy_setopt(easy.raw(), curl_sys::CURLOPT_DEBUGFUNCTION, callback), + curl_sys::CURLE_OK + ); + assert_eq!( + curl_sys::curl_easy_setopt( + easy.raw(), + curl_sys::CURLOPT_DEBUGDATA, + std::ptr::from_mut(counter).cast::(), + ), + curl_sys::CURLE_OK + ); + } +} + +fn read_head(stream: &mut std::net::TcpStream) -> std::io::Result> { + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + let mut head = Vec::new(); + let mut byte = [0]; + while !head.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte)?; + head.extend_from_slice(&byte); + } + Ok(head) +} + +#[test] +fn prerequest_timing_survives_moves_and_reuse_without_verbose() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let request = Request::get(&format!("http://{}/timing", listener.local_addr()?))?; + let server = thread::spawn(move || -> std::io::Result<()> { + // All three requests must reuse one connection, including after reset. + let (mut stream, _) = listener.accept()?; + for method in ["GET", "HEAD", "POST"] { + let head = read_head(&mut stream)?; + assert!(head.starts_with(format!("{method} ").as_bytes())); + if method == "POST" { + let mut body = [0; 5]; + stream.read_exact(&mut body)?; + assert_eq!(&body, b"probe"); + } + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")?; + } + Ok(()) + }); + let cancel = FetchCancelHandle::new(); + let mut debug_calls = Box::new(0); + let mut easy = Easy2::new(FetchTransferHandler::new_buffered(ResponseCollector::new( + Some(cancel.clone()), + ))); + let mut previous = Vec::new(); + for method in ["GET", "HEAD", "POST"] { + easy.reset(); + let before = Instant::now(); + cancel.begin_response_progress(); + let observer = FetchCancelHandle::new(); + observer.adopt_response_progress(&cancel); + easy.get_mut().buffered_mut().unwrap().begin_request(None); + configure_network_observation(&mut easy, &request, None, false)?; + install_debug_counter(&mut easy, &mut debug_calls); + easy.url(request.url.as_str())?; + easy.proxy("")?; + easy.timeout(Duration::from_secs(5))?; + if method == "HEAD" { + easy.nobody(true)?; + } else if method == "POST" { + easy.post_fields_copy(b"probe")?; + } + // A newer selected token must not redirect callbacks on this easy. + cancel.begin_response_progress(); + // Moving Easy2 to another thread must leave PREREQDATA valid. + easy = thread::spawn(move || { + easy.perform().unwrap(); + easy + }) + .join() + .unwrap(); + let timing = observer.response_timing(); + assert!(timing.request_started_at().unwrap() >= before); + assert!(timing.request_started_at().unwrap() <= timing.response_started_at().unwrap()); + assert!(cancel.response_timing().request_started_at().is_none()); + assert_eq!(*debug_calls, 0, "ordinary requests must not enable verbose"); + previous.push((observer, timing)); + for (observer, snapshot) in &previous { + assert_eq!(&observer.response_timing(), snapshot); + } + } + drop(easy); + server.join().unwrap()?; + Ok(()) +} + +#[test] +fn verbose_remains_enabled_for_header_observation_and_proxy_diagnostics() -> Result<()> { + for capture_proxy in [false, true] { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let mut request = Request::get(&format!("http://{}/verbose", listener.local_addr()?))?; + if !capture_proxy { + request = + request.with_network_observation_recorder(NetworkObservationRecorder::default()); + } + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_head(&mut stream)?; + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + }); + let cancel = FetchCancelHandle::new(); + let mut debug_calls = Box::new(0); + let mut easy = Easy2::new(FetchTransferHandler::new_buffered(ResponseCollector::new( + Some(cancel.clone()), + ))); + configure_network_observation(&mut easy, &request, None, capture_proxy)?; + install_debug_counter(&mut easy, &mut debug_calls); + easy.url(request.url.as_str())?; + easy.proxy("")?; + easy.timeout(Duration::from_secs(5))?; + easy.perform()?; + assert!(*debug_calls > 0); + assert!(cancel.response_timing().request_started_at().is_some()); + drop(easy); + server.join().unwrap()?; + } + Ok(()) +} + +#[test] +fn failed_proxy_connect_does_not_start_the_origin_request() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let proxy = format!("http://{}", listener.local_addr()?); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + assert!(read_head(&mut stream)?.starts_with(b"CONNECT ")); + stream + .write_all(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + }); + let request = Request::get("https://origin.invalid/timing")?; + let cancel = FetchCancelHandle::new(); + let mut easy = Easy2::new(FetchTransferHandler::new_buffered(ResponseCollector::new( + Some(cancel.clone()), + ))); + configure_network_observation(&mut easy, &request, None, true)?; + easy.url(request.url.as_str())?; + easy.proxy(&proxy)?; + easy.noproxy("")?; + easy.timeout(Duration::from_secs(5))?; + assert!(easy.perform().is_err()); + assert!(cancel.response_timing().request_started_at().is_none()); + assert_eq!( + take_failed_proxy_connect_response(&mut easy) + .unwrap() + .status, + 403 + ); + drop(easy); + server.join().unwrap()?; + Ok(()) +} diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs new file mode 100644 index 00000000000..dd72c1d130b --- /dev/null +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -0,0 +1,491 @@ +use super::*; +use moli_cookie_jar::new_shared_browser_cookie_store; +use std::{io::Write, sync::mpsc as std_mpsc}; + +fn submit_html_request( + config: &FetchConfig, + request: Request, +) -> ( + FetchRuntimeOwner, + FetchCancelHandle, + PendingStreamingHtmlResponse, +) { + let owner = FetchRuntimeOwner::new(config, new_shared_browser_cookie_store()); + let pending = owner.handle().submit_html_stream(request).unwrap(); + let cancel = pending.cancel_handle.clone(); + (owner, cancel, pending) +} + +#[tokio::test] +async fn html_redirect_timing_belongs_to_the_final_transfer() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/redirect", listener.local_addr()?); + let (release_redirect, ready) = std_mpsc::sync_channel(1); + let server = thread::spawn(move || -> std::io::Result<()> { + for path in ["/redirect", "/final"] { + let (mut stream, _) = listener.accept()?; + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + let mut request = Vec::new(); + let mut byte = [0]; + while !request.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte)?; + request.extend_from_slice(&byte); + } + assert!(request.starts_with(format!("GET {path} ").as_bytes())); + if path == "/redirect" { + stream.write_all( + b"HTTP/1.1 103 Early Hints\r\n\r\n\ + HTTP/1.1 302 Found\r\nLocation: /final\r\nContent-Length: 8\r\nConnection: close\r\n\r\nold!", + )?; + ready.recv_timeout(Duration::from_secs(5)).unwrap(); + stream.write_all(b"tail")?; + } else { + stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + )?; + } + } + Ok(()) + }); + let (_owner, cancel, pending) = + submit_html_request(&FetchConfig::default(), Request::get(&url)?); + let deadline = Instant::now() + Duration::from_secs(5); + while !cancel.response_timing().body_bytes_received() { + anyhow::ensure!(Instant::now() < deadline, "redirect body was not observed"); + tokio::time::sleep(Duration::from_millis(1)).await; + } + let redirect_timing = cancel.response_timing(); + assert!( + redirect_timing + .first_interim_response_started_at() + .is_some() + ); + assert!(redirect_timing.response_started_at().is_some()); + let redirect_finished_after = Instant::now(); + release_redirect.send(())?; + + let mut response = pending.into_response().await?; + assert_eq!(response.final_url.path(), "/final"); + assert!(response.next_chunk().await.is_none()); + response.finish().await?; + let timing = cancel.response_timing(); + assert!(timing.request_started_at().unwrap() >= redirect_finished_after); + assert!(timing.response_started_at().unwrap() >= redirect_finished_after); + assert!(timing.first_interim_response_started_at().is_none()); + assert!(!timing.body_bytes_received()); + server.join().expect("server thread")?; + Ok(()) +} + +#[tokio::test] +async fn html_response_timing_finishes_without_consumer_polling() -> Result<()> { + for (framing, body) in [ + ("Content-Length: 4", b"body".as_slice()), + ( + "Transfer-Encoding: chunked", + b"4\r\nbody\r\n0\r\n\r\n".as_slice(), + ), + ] { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/timed", listener.local_addr()?); + let (release_body, ready) = std_mpsc::sync_channel(1); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + write!( + stream, + "HTTP/1.1 200 OK\r\n{framing}\r\nConnection: close\r\n\r\n" + )?; + ready.recv_timeout(Duration::from_secs(5)).unwrap(); + stream.write_all(body)?; + Ok(()) + }); + let (_owner, cancel, pending) = + submit_html_request(&FetchConfig::default(), Request::get(&url)?); + let mut response = pending.into_response().await?; + let headers = cancel.response_timing(); + assert!(headers.response_started_at().is_some()); + assert!(headers.response_end_at().is_none()); + release_body.send(())?; + // Keep the consumer's executor occupied while native threads finish. + let deadline = Instant::now() + Duration::from_secs(5); + while !cancel.response_completion_is_committed() { + anyhow::ensure!( + Instant::now() < deadline, + "HTML transfer did not record completion" + ); + thread::sleep(Duration::from_millis(1)); + } + let native = cancel.response_timing(); + assert!(native.body_bytes_received()); + assert!(native.response_started_at().unwrap() <= native.response_end_at().unwrap()); + assert!(cancel.response_completion_is_committed()); + assert!(!cancel.cancel_if_response_pending()); + let mut text = String::new(); + while let Some(chunk) = response.next_chunk().await { + text.push_str(&chunk); + } + response.finish().await?; + assert_eq!(text, "body"); + assert_eq!(cancel.response_timing(), native); + server.join().expect("server thread")?; + } + Ok(()) +} + +fn read_request(stream: &mut std::net::TcpStream) -> std::io::Result<()> { + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + let mut request = Vec::new(); + let mut byte = [0]; + while !request.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte)?; + request.extend_from_slice(&byte); + } + Ok(()) +} + +#[tokio::test] +async fn html_response_timing_records_terminal_failures() -> Result<()> { + for bytes in [ + b"".as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\nConnection: close\r\n\r\npart".as_slice(), + ] { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/failure", listener.local_addr()?); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + stream.write_all(bytes)?; + Ok(()) + }); + let (_owner, cancel, pending) = + submit_html_request(&FetchConfig::default(), Request::get(&url)?); + match pending.into_response().await { + Ok(mut response) => assert!(response.finish().await.is_err()), + Err(_) => assert!(bytes.is_empty()), + } + let timing = cancel.response_timing(); + assert_eq!(timing.response_started_at().is_some(), !bytes.is_empty()); + assert!(timing.response_end_at().is_some()); + assert!(cancel.response_completion_is_committed()); + assert!(!cancel.cancel_if_response_pending()); + server.join().expect("server thread")?; + } + Ok(()) +} + +#[tokio::test] +async fn html_response_timing_finishes_bodyless_responses_at_headers() -> Result<()> { + for (method, headers) in [ + ("GET", "HTTP/1.1 101 Switching Protocols\r\n"), + ("GET", "HTTP/1.1 204 No Content\r\n"), + ("GET", "HTTP/1.1 205 Reset Content\r\n"), + ("GET", "HTTP/1.1 200 OK\r\nContent-Length: 0\r\n"), + ("HEAD", "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n"), + ] { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/empty", listener.local_addr()?); + let (release, ready) = std_mpsc::sync_channel(1); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + write!(stream, "{headers}Connection: close\r\n\r\n")?; + ready.recv_timeout(Duration::from_secs(5)).unwrap(); + Ok(()) + }); + let request = Request::new(method, &url, None, Vec::new())?; + let (_owner, cancel, pending) = submit_html_request(&FetchConfig::default(), request); + let mut response = pending.into_response().await?; + tokio::time::timeout(Duration::from_secs(5), response.finish()).await??; + let timing = cancel.response_timing(); + assert!(timing.response_end_at().is_some(), "{method}: {headers}"); + assert!(timing.response_started_at().unwrap() <= timing.response_end_at().unwrap()); + assert!(!timing.body_bytes_received()); + assert!(!cancel.cancel_if_response_pending()); + release.send(())?; + server.join().expect("server thread")?; + } + Ok(()) +} + +#[tokio::test] +async fn html_response_timing_preserves_cancellation_before_body_completion() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/cancel", listener.local_addr()?); + let (release, ready) = std_mpsc::sync_channel(1); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\nConnection: close\r\n\r\npart")?; + ready.recv_timeout(Duration::from_secs(5)).unwrap(); + Ok(()) + }); + let (_owner, cancel, pending) = + submit_html_request(&FetchConfig::default(), Request::get(&url)?); + let mut response = pending.into_response().await?; + assert_eq!(response.next_chunk().await.as_deref(), Some("part")); + assert!(cancel.response_timing().response_end_at().is_none()); + assert!(cancel.cancel_if_response_pending()); + assert!( + tokio::time::timeout(Duration::from_secs(5), response.finish()) + .await? + .is_err() + ); + assert!(cancel.response_timing().response_end_at().is_some()); + assert!(!cancel.response_completion_is_committed()); + release.send(())?; + server.join().expect("server thread")?; + Ok(()) +} + +#[tokio::test] +async fn html_cached_response_timing_does_not_invent_network_observations() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let url = format!("http://{}/cache", listener.local_addr()?); + let cache_dir = std::env::temp_dir().join(format!( + "moli-html-timing-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_nanos(), + )); + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + stream.write_all(b"HTTP/1.1 200 OK\r\nCache-Control: max-age=600\r\nContent-Length: 4\r\nConnection: close\r\n\r\nbody")?; + Ok(()) + }); + let mut config = FetchConfig::default(); + config.set_http_cache_dir(Some(cache_dir.display().to_string())); + let (owner, _, pending) = submit_html_request(&config, Request::get(&url)?); + pending.into_response().await?.finish().await?; + server.join().expect("server thread")?; + + let pending = owner.handle().submit_html_stream(Request::get(&url)?)?; + let cancel = pending.cancel_handle.clone(); + let mut response = pending.into_response().await?; + assert!(response.from_cache); + let mut text = String::new(); + while let Some(chunk) = response.next_chunk().await { + text.push_str(&chunk); + } + response.finish().await?; + assert_eq!(text, "body"); + let timing = cancel.response_timing(); + assert!(timing.request_started_at().is_none()); + assert!(timing.first_interim_response_started_at().is_none()); + assert!(!timing.body_bytes_received()); + assert!(timing.response_started_at().is_some()); + assert!(timing.response_started_at().unwrap() <= timing.response_end_at().unwrap()); + assert!(!cancel.cancel_if_response_pending()); + drop(owner); + std::fs::remove_dir_all(cache_dir)?; + Ok(()) +} + +#[tokio::test] +async fn html_declared_body_completion_precedes_delivery_of_the_last_chunk() -> Result<()> { + for (encoding_headers, complete) in [ + (b"".as_slice(), true), + (b"Content-Encoding: gzip\r\n".as_slice(), false), + (b"Transfer-Encoding: chunked\r\n".as_slice(), false), + ] { + let cancel = FetchCancelHandle::new(); + let (start_tx, _start_rx) = oneshot::channel(); + let (body_tx, mut body_rx) = mpsc::unbounded_channel(); + let mut collector = StreamingResponseCollector::new( + new_shared_browser_cookie_store(), + start_tx, + body_tx, + cancel.clone(), + ); + collector.begin_request( + None, + Url::parse("http://example.test/body")?, + NetworkCookieRequestContext::top_level_navigation("GET"), + None, + true, + Vec::new(), + None, + ); + let mut handler = FetchTransferHandler::new_streaming(collector); + assert!(handler.header(b"HTTP/1.1 200 OK\r\n")); + assert!(handler.header(b"Content-Length: 4\r\n")); + if !encoding_headers.is_empty() { + assert!(handler.header(encoding_headers)); + } + assert!(handler.header(b"\r\n")); + assert_eq!(handler.write(b"bo").unwrap(), 2); + assert_eq!(body_rx.try_recv()?.as_str(), "bo"); + assert!(cancel.response_timing().body_bytes_received()); + assert!(cancel.response_timing().response_end_at().is_none()); + assert_eq!(handler.write(b"dy").unwrap(), 2); + assert_eq!(body_rx.try_recv()?.as_str(), "dy"); + // Decoded or chunked lengths cannot be compared with Content-Length. + assert_eq!( + cancel.response_timing().response_end_at().is_some(), + complete + ); + assert_eq!(cancel.cancel_if_response_pending(), !complete); + } + Ok(()) +} + +#[test] +fn switching_protocols_timing_is_final_in_every_collector() { + for interim in [false, true] { + for mode in ["buffered", "html", "raw"] { + let cancel = FetchCancelHandle::new(); + let mut handler = match mode { + "buffered" => { + FetchTransferHandler::new_buffered(ResponseCollector::new(Some(cancel.clone()))) + } + "html" => { + let (start_tx, _start_rx) = oneshot::channel(); + let (body_tx, _body_rx) = mpsc::unbounded_channel(); + FetchTransferHandler::new_streaming(StreamingResponseCollector::new( + new_shared_browser_cookie_store(), + start_tx, + body_tx, + cancel.clone(), + )) + } + "raw" => { + let (start_tx, _start_rx) = oneshot::channel(); + let (body_tx, _body_rx) = mpsc::unbounded_channel(); + FetchTransferHandler::new_raw_streaming(RawStreamingResponseCollector::new( + new_shared_browser_cookie_store(), + start_tx, + body_tx, + cancel.clone(), + )) + } + _ => unreachable!(), + }; + if interim { + assert!(handler.header(b"HTTP/1.1 103 Early Hints\r\n")); + assert!(cancel.response_timing().response_started_at().is_none()); + } + assert!(handler.header(b"HTTP/1.1 101 Switching Protocols\r\n")); + let timing = cancel.response_timing(); + assert!( + timing.response_started_at().is_some(), + "{mode}: 101 must be final" + ); + assert_eq!( + timing.first_interim_response_started_at().is_some(), + interim, + "{mode}" + ); + if let Some(first_interim) = timing.first_interim_response_started_at() { + assert!(first_interim <= timing.response_started_at().unwrap()); + } + } + } +} + +#[test] +fn transfer_progress_preserves_cancellation_and_body_observations_on_reuse() { + let cancel = FetchCancelHandle::new(); + let mut handler = + FetchTransferHandler::new_buffered(ResponseCollector::new(Some(cancel.clone()))); + for _ in 0..2 { + cancel.begin_response_progress(); + handler.buffered_mut().unwrap().begin_request(None); + handler.begin_transfer(None, false); + assert!(handler.progress(4.0, 0.0, 0.0, 0.0)); + assert!(!cancel.response_timing().body_bytes_received()); + assert!(handler.progress(4.0, 1.0, 0.0, 0.0)); + assert!(cancel.response_timing().body_bytes_received()); + assert!(handler.progress(4.0, 4.0, 0.0, 0.0)); + } + cancel.cancel(); + assert!(!handler.progress(4.0, 4.0, 0.0, 0.0)); +} + +#[tokio::test] +async fn html_redirect_to_a_cached_response_preserves_channels_and_timing() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = format!("http://{}", listener.local_addr()?); + let cache_dir = std::env::temp_dir().join(format!( + "moli-html-redirect-timing-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_nanos(), + )); + let server = thread::spawn(move || -> std::io::Result<()> { + for response in [ + b"HTTP/1.1 200 OK\r\nCache-Control: max-age=600\r\nContent-Length: 4\r\nConnection: close\r\n\r\nbody".as_slice(), + b"HTTP/1.1 103 Early Hints\r\n\r\nHTTP/1.1 302 Found\r\nLocation: /cache\r\nContent-Length: 4\r\nConnection: close\r\n\r\njunk".as_slice(), + ] { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + stream.write_all(response)?; + } + Ok(()) + }); + let mut config = FetchConfig::default(); + config.set_http_cache_dir(Some(cache_dir.display().to_string())); + let (owner, _, pending) = + submit_html_request(&config, Request::get(&format!("{address}/cache"))?); + pending.into_response().await?.finish().await?; + + let pending = owner + .handle() + .submit_html_stream(Request::get(&format!("{address}/redirect"))?)?; + let cancel = pending.cancel_handle.clone(); + let mut response = pending.into_response().await?; + assert!(response.from_cache); + assert!(response.redirected); + assert_eq!(response.final_url.path(), "/cache"); + assert_eq!(response.redirect_chain.len(), 1); + assert!(!response.redirect_chain[0].from_cache); + let mut body = String::new(); + while let Some(chunk) = response.next_chunk().await { + body.push_str(&chunk); + } + response.finish().await?; + assert_eq!(body, "body"); + let timing = cancel.response_timing(); + assert!(timing.request_started_at().is_none()); + assert!(timing.first_interim_response_started_at().is_none()); + assert!(!timing.body_bytes_received()); + assert!(timing.response_started_at().unwrap() <= timing.response_end_at().unwrap()); + assert!(!cancel.cancel_if_response_pending()); + server.join().expect("server thread")?; + drop(owner); + std::fs::remove_dir_all(cache_dir)?; + Ok(()) +} + +#[tokio::test] +async fn html_failure_before_reusing_an_easy_preserves_the_original_error() -> Result<()> { + let cancel = FetchCancelHandle::new(); + let (started_tx, started_rx) = oneshot::channel(); + let (body_tx, mut body_rx) = mpsc::unbounded_channel(); + let (completion_tx, completion_rx) = oneshot::channel(); + let mut job = StreamingRuntimeJob::new( + Request::get("http://example.test/cache")?, + started_tx, + body_tx, + completion_tx, + cancel.clone(), + ); + job.easy = Some(Easy2::new(FetchTransferHandler::new_streaming( + StreamingResponseCollector::new( + new_shared_browser_cookie_store(), + job.started_tx.take().unwrap(), + job.body_tx.take().unwrap(), + cancel.clone(), + ), + ))); + fail_streaming_job_with_easy(job, None, anyhow!("cached redirect rejected")); + let error = started_rx.await?.unwrap_err(); + assert!(format!("{error:#}").contains("cached redirect rejected")); + assert!(body_rx.recv().await.is_none()); + assert!(completion_rx.await?.is_err()); + assert!(cancel.response_timing().response_end_at().is_some()); + Ok(()) +} diff --git a/moli-fetch/src/tests/response_timing.rs b/moli-fetch/src/tests/response_timing.rs index d2e1172ead1..3d650360a4b 100644 --- a/moli-fetch/src/tests/response_timing.rs +++ b/moli-fetch/src/tests/response_timing.rs @@ -153,3 +153,138 @@ async fn raw_response_timing_is_native_even_when_the_consumer_does_not_poll() -> server.join().expect("server thread")?; Ok(()) } + +#[tokio::test] +async fn native_request_timing_is_transfer_scoped_on_multiplexed_http2() -> Result<()> { + let server = ScriptedH2Server::spawn_with_response_barrier( + vec![ScriptedResponse::ok("h2"); 2], + Arc::new(tokio::sync::Barrier::new(2)), + ); + let mut config = FetchConfig::default(); + config.set_tls_verify_host(false); + config.set_transport_connection_limits(Some(1), Some(1), Some(2)); + let owner = FetchRuntimeOwner::new(&config, new_shared_browser_cookie_store()); + let handle = owner.handle(); + let buffered_cancel = FetchCancelHandle::new(); + let raw_cancel = FetchCancelHandle::new(); + let before = Instant::now(); + let buffered = handle.submit_with_cancel( + Request::get(&server.url_path("/buffered"))?, + buffered_cancel.clone(), + )?; + let raw = + handle.submit_raw_stream(Request::get(&server.url_path("/raw"))?, raw_cancel.clone())?; + let requests = async { + let buffered = async { + let response = buffered.await??; + assert_eq!( + response.negotiated_http_version, + Some(NegotiatedHttpVersion::Http2) + ); + assert_eq!(response.body_text(), "h2"); + Ok::<_, anyhow::Error>(()) + }; + let raw = async { + let mut response = raw.into_response().await?; + assert_eq!( + response.negotiated_http_version, + Some(NegotiatedHttpVersion::Http2) + ); + let mut body = Vec::new(); + while let Some(chunk) = response.next_chunk().await { + body.extend_from_slice(&chunk); + } + response.finish().await?; + assert_eq!(body, b"h2"); + Ok::<_, anyhow::Error>(()) + }; + tokio::try_join!(buffered, raw) + }; + tokio::time::timeout(Duration::from_secs(5), requests).await??; + let after = Instant::now(); + for cancel in [buffered_cancel, raw_cancel] { + let timing = cancel.response_timing(); + assert!(timing.request_started_at().unwrap() >= before); + assert!(timing.request_started_at().unwrap() <= timing.response_started_at().unwrap()); + assert!(timing.response_started_at().unwrap() <= timing.response_end_at().unwrap()); + assert!(timing.response_end_at().unwrap() <= after); + } + // The barrier withholds all responses until both streams have arrived. + // A single connection therefore proves concurrent multiplexing, not just + // sequential reuse of an HTTP/2 connection. + assert_eq!(server.connection_stream_counts(), [2]); + server.shutdown(); + Ok(()) +} + +#[tokio::test] +async fn native_request_timing_waits_for_the_proxy_tunnel() -> Result<()> { + let server = ScriptedH2Server::spawn(vec![ScriptedResponse::ok("tunneled")]); + let url = server.url_path("/timing"); + let origin = Url::parse(&url)?; + let origin_port = origin.port().unwrap(); + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + listener.set_nonblocking(true)?; + let proxy_url = format!("http://{}", listener.local_addr()?); + let (connect_seen_tx, connect_seen) = oneshot::channel(); + let (release_connect, release_connect_rx) = oneshot::channel(); + let proxy = thread::spawn(move || -> Result<()> { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(async { + let listener = TcpListener::from_std(listener)?; + let (mut downstream, _) = listener.accept().await?; + let mut head = Vec::new(); + let mut byte = [0]; + while !head.ends_with(b"\r\n\r\n") { + downstream.read_exact(&mut byte).await?; + head.extend_from_slice(&byte); + } + assert!(head.starts_with(b"CONNECT ")); + connect_seen_tx.send(()).unwrap(); + release_connect_rx.await?; + let mut upstream = + tokio::net::TcpStream::connect(("127.0.0.1", origin_port)).await?; + downstream + .write_all(b"HTTP/1.1 200 Connection established\r\n\r\n") + .await?; + // Dropping the client after completion closes the tunnel. + let _ = tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await; + Ok(()) + }) + }); + let mut config = FetchConfig::default(); + config.set_tls_verify_host(false); + config.set_http_proxy(Some(proxy_url)); + let client = FetchClient::new(&config, new_shared_browser_cookie_store()); + let cancel = FetchCancelHandle::new(); + let fetch = client.fetch_raw_stream_with_cancel(Request::get(&url)?, cancel.clone()); + let release = async { + tokio::time::timeout(Duration::from_secs(5), connect_seen).await??; + assert!(cancel.response_timing().request_started_at().is_none()); + let released_at = Instant::now(); + release_connect.send(()).unwrap(); + Ok::<_, anyhow::Error>(released_at) + }; + let (response, released_at) = tokio::join!(fetch, release); + let mut response = response?; + assert_eq!( + response.negotiated_http_version, + Some(NegotiatedHttpVersion::Http2) + ); + let mut body = Vec::new(); + while let Some(chunk) = response.next_chunk().await { + body.extend_from_slice(&chunk); + } + response.finish().await?; + assert_eq!(body, b"tunneled"); + let timing = cancel.response_timing(); + assert!(timing.request_started_at().unwrap() >= released_at?); + assert!(timing.request_started_at().unwrap() <= timing.response_started_at().unwrap()); + drop(response); + drop(client); + proxy.join().unwrap()?; + server.shutdown(); + Ok(()) +} diff --git a/moli-fetch/src/tests/support.rs b/moli-fetch/src/tests/support.rs index 92f5c7b093c..a5d61d9a727 100644 --- a/moli-fetch/src/tests/support.rs +++ b/moli-fetch/src/tests/support.rs @@ -237,6 +237,20 @@ impl ScriptedHttps11Server { impl ScriptedH2Server { pub(super) fn spawn(responses: Vec) -> Self { + Self::spawn_with_gate(responses, None) + } + + pub(super) fn spawn_with_response_barrier( + responses: Vec, + barrier: Arc, + ) -> Self { + Self::spawn_with_gate(responses, Some(barrier)) + } + + fn spawn_with_gate( + responses: Vec, + barrier: Option>, + ) -> Self { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); listener .set_nonblocking(true) @@ -258,6 +272,7 @@ impl ScriptedH2Server { requests_for_thread, connection_stream_counts_for_thread, responses, + barrier, ); }); @@ -416,6 +431,7 @@ fn run_h2_server( requests: Arc>>, connection_stream_counts: Arc>>, responses: Arc>>, + barrier: Option>, ) { let rt = tokio::runtime::Builder::new_current_thread() .enable_io() @@ -438,6 +454,7 @@ fn run_h2_server( let requests = Arc::clone(&requests); let connection_stream_counts = Arc::clone(&connection_stream_counts); let responses = Arc::clone(&responses); + let barrier = barrier.clone(); tokio::spawn(async move { let Ok(stream) = tls_acceptor.accept(stream).await else { return; @@ -470,7 +487,11 @@ fn run_h2_server( .or_else(|| responses.back().cloned()) .expect("scripted h2 responses should not be empty") }; + let barrier = barrier.clone(); tokio::spawn(async move { + if let Some(barrier) = barrier { + barrier.wait().await; + } if response_spec.delay_ms > 0 { tokio::time::sleep(Duration::from_millis(response_spec.delay_ms)).await; }