From ad64f266e68c5966ddee7b572e75304a0d265a0d Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 1/9] fix(fetch): isolate HTML timing for each transfer attempt Select a fresh ResponseProgress for every HTML attempt and pass that exact writer token to the collector, matching raw streaming. Redirects and retries must not retain observations from a discarded hop. Add a gated loopback 103 -> 302 with body -> 200 without body regression. --- moli-fetch/src/blocking/collectors.rs | 2 + moli-fetch/src/runtime.rs | 5 ++ .../src/runtime/response_timing_tests.rs | 78 +++++++++++++++++++ 3 files changed, 85 insertions(+) create mode 100644 moli-fetch/src/runtime/response_timing_tests.rs diff --git a/moli-fetch/src/blocking/collectors.rs b/moli-fetch/src/blocking/collectors.rs index fcecd394969..b2dd3ad2810 100644 --- a/moli-fetch/src/blocking/collectors.rs +++ b/moli-fetch/src/blocking/collectors.rs @@ -369,6 +369,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 +380,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; } diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index d13cab77aa1..d7f60997317 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); @@ -3790,6 +3792,9 @@ fn request_panics_for_testing(request: &Request) -> bool { }) } +#[cfg(test)] +mod response_timing_tests; + #[cfg(test)] mod tests { use super::*; 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..176b684dc90 --- /dev/null +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -0,0 +1,78 @@ +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(()) +} From f2e8b997a14087ca1874f9ee197be83d891f540f Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 2/9] fix(fetch): record native terminal timing for HTML streams Record response end before publishing native success or failure, including cancellation, header-only responses and declared identity body completion. Keep cache availability separate from the end of the cached body reader. Cover delayed consumers, chunked bodies, empty responses, truncated transfers, cancellation, cache hits and encoded Content-Length comparison boundaries. Observe delivered body bytes before publishing declared completion. --- moli-fetch/src/blocking/collectors.rs | 14 + moli-fetch/src/runtime.rs | 10 + .../src/runtime/response_timing_tests.rs | 252 ++++++++++++++++++ 3 files changed, 276 insertions(+) diff --git a/moli-fetch/src/blocking/collectors.rs b/moli-fetch/src/blocking/collectors.rs index b2dd3ad2810..acc40d2f6f2 100644 --- a/moli-fetch/src/blocking/collectors.rs +++ b/moli-fetch/src/blocking/collectors.rs @@ -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, @@ -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; @@ -503,6 +506,7 @@ impl StreamingResponseCollector { self.client_hint_restart_requested = true; return false; } + self.declared_identity_body_length = identity_encoded_content_length(&self.headers); // Create the cache writer only after final response headers prove the // response is cacheable. self.maybe_create_cache_body_writer(); @@ -548,6 +552,11 @@ impl StreamingResponseCollector { { return; } + if self.declared_identity_body_length == Some(0) + || 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 { @@ -1116,6 +1125,10 @@ impl Handler for StreamingResponseCollector { } self.response_bytes_received += data.len(); + if self.started && self.declared_identity_body_length == Some(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() { @@ -1139,6 +1152,7 @@ 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 { diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index d7f60997317..5553b91167c 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -1928,6 +1928,7 @@ impl RuntimeOwner { return; } + job.response_progress.commit(); let cache_body_writer = { let streaming = easy .get_mut() @@ -2828,6 +2829,11 @@ impl FetchTransferHandler { impl Handler for FetchTransferHandler { fn write(&mut self, data: &[u8]) -> std::result::Result { + if !data.is_empty() + && let Some(timing) = self.response_progress() + { + timing.mark_response_body_received(); + } match &mut self.response { FetchResponseCollector::Buffered(collector) => collector.write(data), FetchResponseCollector::Streaming(collector) => collector.write(data), @@ -2981,6 +2987,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}" @@ -2993,6 +3000,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, @@ -3014,6 +3022,7 @@ fn complete_cached_streaming_html_job( } else { Ok(()) }; + response_timing.finish(); let _ = job.completion_tx.send(completion); } @@ -3698,6 +3707,7 @@ fn fail_streaming_job_with_easy( &job.request.redirect_chain, error, ); + job.response_progress.commit(); if let Some(easy) = easy.as_mut() && let Some(streaming) = easy.get_mut().streaming_mut() { diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs index 176b684dc90..ac404bd75bf 100644 --- a/moli-fetch/src/runtime/response_timing_tests.rs +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -76,3 +76,255 @@ async fn html_redirect_timing_belongs_to_the_final_transfer() -> Result<()> { 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 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, + ); + assert!(collector.header(b"HTTP/1.1 200 OK\r\n")); + assert!(collector.header(b"Content-Length: 4\r\n")); + if !encoding_headers.is_empty() { + assert!(collector.header(encoding_headers)); + } + assert!(collector.header(b"\r\n")); + assert_eq!(collector.write(b"bo").unwrap(), 2); + assert_eq!(body_rx.try_recv()?.as_str(), "bo"); + assert!(cancel.response_timing().response_end_at().is_none()); + assert_eq!(collector.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(()) +} From 10091de3928fbd70951531c0600b2c46f9d99da7 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 3/9] fix(fetch): classify 101 as final in response timing Reuse the existing interim status predicate in buffered, HTML and raw collectors so Switching Protocols starts the final response timeline. Cover 101 alone and after 103 across all collectors, plus HTML header completion. --- moli-fetch/src/blocking/collectors.rs | 10 ++-- .../src/runtime/response_timing_tests.rs | 54 +++++++++++++++++++ 2 files changed, 59 insertions(+), 5 deletions(-) diff --git a/moli-fetch/src/blocking/collectors.rs b/moli-fetch/src/blocking/collectors.rs index acc40d2f6f2..6db422d7007 100644 --- a/moli-fetch/src/blocking/collectors.rs +++ b/moli-fetch/src/blocking/collectors.rs @@ -1078,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(); @@ -1155,7 +1155,7 @@ impl Handler for StreamingResponseCollector { 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(); @@ -1233,10 +1233,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; } diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs index ac404bd75bf..f122732ce47 100644 --- a/moli-fetch/src/runtime/response_timing_tests.rs +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -177,6 +177,7 @@ async fn html_response_timing_records_terminal_failures() -> Result<()> { #[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"), @@ -328,3 +329,56 @@ async fn html_declared_body_completion_precedes_delivery_of_the_last_chunk() -> } 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()); + } + } + } +} From 653320ee319c8693ca8d2871411c59704468c147 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 4/9] perf(fetch): share transfer progress state in one allocation Keep cancellation arbitration and timing in one Arc. This removes one allocation per progress generation and one reference-count update per clone, reducing the writer token from 16 to 8 bytes on x86_64. Preserve transfer isolation and the existing atomic completion/cancel protocol. --- moli-fetch/src/cancellation.rs | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/moli-fetch/src/cancellation.rs b/moli-fetch/src/cancellation.rs index 4cf4838c794..fe9c50c73f2 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, @@ -226,7 +235,7 @@ impl FetchCancelHandle { } pub fn response_timing(&self) -> FetchResponseTiming { - *self.response_progress().timing.lock() + *self.response_progress().inner.timing.lock() } #[cfg(test)] From b0d0c3a2737737727720c90d5921fe500cb0bcd4 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 5/9] perf(fetch): record response body presence once per transfer Latch the first body observation in the transfer handler and reset it whenever the easy handle is reused. Record nonempty writes before declared completion, while retaining progress-based detection for compressed empty payloads. Avoid repeated timing locks without skipping cancellation checks. --- moli-fetch/src/runtime.rs | 24 +++++++++----- .../src/runtime/response_timing_tests.rs | 33 +++++++++++++++---- 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index 5553b91167c..a07e04ff7ce 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -2720,6 +2720,7 @@ struct ActiveRawStreamingTransferContext { pub(crate) struct FetchTransferHandler { response: FetchResponseCollector, + response_body_observed: bool, network_observation_recorder: Option, proxy_connect_response_recorder: ProxyConnectResponseRecorder, } @@ -2731,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(), @@ -2754,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(), } @@ -2806,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); @@ -2829,10 +2841,8 @@ impl FetchTransferHandler { impl Handler for FetchTransferHandler { fn write(&mut self, data: &[u8]) -> std::result::Result { - if !data.is_empty() - && let Some(timing) = self.response_progress() - { - timing.mark_response_body_received(); + if !data.is_empty() { + self.mark_response_body_received(); } match &mut self.response { FetchResponseCollector::Buffered(collector) => collector.write(data), @@ -2853,10 +2863,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) => { diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs index f122732ce47..4e60d3e0b59 100644 --- a/moli-fetch/src/runtime/response_timing_tests.rs +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -309,16 +309,18 @@ async fn html_declared_body_completion_precedes_delivery_of_the_last_chunk() -> Vec::new(), None, ); - assert!(collector.header(b"HTTP/1.1 200 OK\r\n")); - assert!(collector.header(b"Content-Length: 4\r\n")); + 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!(collector.header(encoding_headers)); + assert!(handler.header(encoding_headers)); } - assert!(collector.header(b"\r\n")); - assert_eq!(collector.write(b"bo").unwrap(), 2); + 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!(collector.write(b"dy").unwrap(), 2); + 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!( @@ -382,3 +384,22 @@ fn switching_protocols_timing_is_final_in_every_collector() { } } } + +#[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)); +} From 26cbeffff038c3a0f224c4da8f1b2017997d4847 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH 6/9] fix(fetch): recover HTML response channels for cached redirects Redirects and retries retain start/body senders in the reused easy handle. Return those channels to the HTML job before publishing a final cache hit, and use the retained easy when an attempt fails before reconfiguration. Cover network-to-cache redirects with discarded interim/body observations, and preserve the original error instead of reporting a closed start channel. --- moli-fetch/src/runtime.rs | 16 +++- .../src/runtime/response_timing_tests.rs | 86 +++++++++++++++++++ 2 files changed, 99 insertions(+), 3 deletions(-) diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index a07e04ff7ce..d2d1e4e6ec9 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -2978,10 +2978,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 { @@ -3705,10 +3714,11 @@ 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, diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs index 4e60d3e0b59..dd72c1d130b 100644 --- a/moli-fetch/src/runtime/response_timing_tests.rs +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -403,3 +403,89 @@ fn transfer_progress_preserves_cancellation_and_body_observations_on_reuse() { 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(()) +} From 3c04b1c46582ee6aa23594eeed1680da6f64c92e Mon Sep 17 00:00:00 2001 From: ldm0 Date: Mon, 19 Oct 2026 00:12:14 +0800 Subject: [PATCH 7/9] perf(fetch): compact declared body lengths with NonZeroUsize Keep zero-length completion local to header processing and retain only positive body-length targets in the HTML and raw streaming collectors. Preserve HTML redirect eligibility before committing an empty response. On the same x86_64 dev build, the field shrinks from 16 to 8 bytes, HTML collectors from 2752 to 2744 bytes, and raw collectors from 2728 to 2720 bytes. The enclosing transfer handler also shrinks by 8 bytes. Cover accepted empty responses, unknown lengths and followed empty redirects without changing their completion/cancellation behavior. --- moli-fetch/src/blocking/collectors.rs | 75 ++++++++++++++++++++++----- 1 file changed, 62 insertions(+), 13 deletions(-) diff --git a/moli-fetch/src/blocking/collectors.rs b/moli-fetch/src/blocking/collectors.rs index 6db422d7007..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,7 +186,7 @@ pub struct StreamingResponseCollector { 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, @@ -258,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, @@ -506,11 +506,12 @@ impl StreamingResponseCollector { self.client_hint_restart_requested = true; return false; } - self.declared_identity_body_length = identity_encoded_content_length(&self.headers); + 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; @@ -538,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; } @@ -552,9 +553,7 @@ impl StreamingResponseCollector { { return; } - if self.declared_identity_body_length == Some(0) - || response_body_ends_at_headers(self.status) - { + if body_is_empty || response_body_ends_at_headers(self.status) { self.response_progress.commit(); } self.started = true; @@ -910,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(); @@ -1125,7 +1125,10 @@ impl Handler for StreamingResponseCollector { } self.response_bytes_received += data.len(); - if self.started && self.declared_identity_body_length == Some(self.response_bytes_received) + if self.started + && self + .declared_identity_body_length + .is_some_and(|length| length.get() == self.response_bytes_received) { self.response_progress.commit(); } @@ -1205,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); @@ -1359,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(); From 9f6055f5b6b0ab288c7317ab2fece675aa96afa6 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Mon, 19 Oct 2026 01:34:12 +0800 Subject: [PATCH 8/9] perf(fetch): observe request start without verbose Record request start with CURLOPT_PREREQFUNCTION using the handler already owned by Easy2. Reinstall the callback on each attempt and read the collector's writer token across moves, reset and progress reselection. Enable verbose only for requested header observations or proxy CONNECT diagnostics. Cover keepalive GET/HEAD/POST reuse, callback ownership, HTTP/2 multiplexing and gated proxy tunnels. --- moli-fetch/src/cancellation.rs | 2 + moli-fetch/src/runtime.rs | 13 +- moli-fetch/src/runtime/request_timing.rs | 54 ++++++ .../src/runtime/request_timing_tests.rs | 182 ++++++++++++++++++ moli-fetch/src/tests/response_timing.rs | 135 +++++++++++++ moli-fetch/src/tests/support.rs | 21 ++ 6 files changed, 401 insertions(+), 6 deletions(-) create mode 100644 moli-fetch/src/runtime/request_timing.rs create mode 100644 moli-fetch/src/runtime/request_timing_tests.rs diff --git a/moli-fetch/src/cancellation.rs b/moli-fetch/src/cancellation.rs index fe9c50c73f2..cb975c29bbe 100644 --- a/moli-fetch/src/cancellation.rs +++ b/moli-fetch/src/cancellation.rs @@ -111,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. Absent for responses served from cache. pub fn request_started_at(self) -> Option { self.request_started_at } diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index d2d1e4e6ec9..223b7ad6b31 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -61,6 +61,8 @@ use crate::{ proxy_connect::{ProxyConnectResponse, ProxyConnectResponseRecorder}, }; +mod request_timing; + const DEFAULT_RUNTIME_TRANSFERS: usize = 256; const RUNTIME_POLL_INTERVAL: Duration = Duration::from_millis(50); // curl-sys does not currently expose CURLINFO_HTTP_VERSION. This value is @@ -2885,9 +2887,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() { @@ -2909,14 +2908,13 @@ 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); + request_timing::configure(easy)?; easy.verbose(verbose) .context("failed to configure curl network observation") } @@ -3823,6 +3821,9 @@ 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.rs b/moli-fetch/src/runtime/request_timing.rs new file mode 100644 index 00000000000..358d17110b1 --- /dev/null +++ b/moli-fetch/src/runtime/request_timing.rs @@ -0,0 +1,54 @@ +use std::{ + ffi::{c_char, c_int, c_void}, + panic::{AssertUnwindSafe, catch_unwind}, +}; + +use super::*; + +// curl-sys does not expose these options yet. Both values are part of curl's +// public curl.h ABI, available since libcurl 7.80.0. +const CURLOPT_PREREQFUNCTION: curl_sys::CURLoption = curl_sys::CURLOPTTYPE_FUNCTIONPOINT + 312; +const CURLOPT_PREREQDATA: curl_sys::CURLoption = curl_sys::CURLOPTTYPE_OBJECTPOINT + 313; + +pub(super) fn configure(easy: &mut Easy2) -> Result<()> { + let callback: extern "C" fn(*mut c_void, *mut c_char, *mut c_char, c_int, c_int) -> c_int = + request_started; + let handler = std::ptr::from_mut(easy.get_mut()).cast::(); + // SAFETY: curl-rust keeps the handler in Easy2's boxed Inner allocation. + // Moving Easy2 does not move the handler, and curl_easy_cleanup runs before + // that allocation is freed. libcurl calls this synchronously on the same + // thread as the other Handler callbacks, with no overlapping handler access. + // Every attempt reconfigures these options after Easy2::reset; its collector + // is bound to that attempt's ResponseProgress before the easy is submitted. + let code = unsafe { curl_sys::curl_easy_setopt(easy.raw(), CURLOPT_PREREQDATA, handler) }; + if code != curl_sys::CURLE_OK { + return Err(curl::Error::new(code)).context("failed to configure curl pre-request data"); + } + // SAFETY: callback has the curl_prereq_callback signature from curl.h. + let code = unsafe { curl_sys::curl_easy_setopt(easy.raw(), CURLOPT_PREREQFUNCTION, callback) }; + if code != curl_sys::CURLE_OK { + return Err(curl::Error::new(code)) + .context("failed to configure curl pre-request callback"); + } + Ok(()) +} + +extern "C" fn request_started( + handler: *mut c_void, + _remote_ip: *mut c_char, + _local_ip: *mut c_char, + _remote_port: c_int, + _local_port: c_int, +) -> c_int { + // Do not unwind across the C ABI. Abort this transfer if the callback fails. + catch_unwind(AssertUnwindSafe(|| { + // SAFETY: configure installs a live FetchTransferHandler as PREREQDATA; + // libcurl invokes callbacks serially while its Easy2 remains alive. + let handler = unsafe { &*handler.cast::() }; + if let Some(timing) = handler.response_progress() { + timing.mark_request_started(); + } + 0 // CURL_PREREQFUNC_OK + })) + .unwrap_or(1) // CURL_PREREQFUNC_ABORT +} 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..e17def320e5 --- /dev/null +++ b/moli-fetch/src/runtime/request_timing_tests.rs @@ -0,0 +1,182 @@ +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_reuse_reset_moves_and_progress_reselection_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/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; } From b393f4723e31b7f6fe1e5ecf73462cc3087e7a23 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Mon, 19 Oct 2026 02:05:05 +0800 Subject: [PATCH 9/9] refactor(fetch): use curl-rust pre-request callbacks Pin curl and curl-sys to the native callback API from curl-rust#1 and replace the handwritten callback registration with Handler::prereq / Easy2::prereq. Keep each collector's captured timing token and re-enable the callback after reset; the dependency owns userdata lifetime and C panic handling. --- Cargo.lock | 4 +- Cargo.toml | 4 +- moli-fetch/src/cancellation.rs | 2 +- moli-fetch/src/runtime.rs | 18 +++++-- moli-fetch/src/runtime/request_timing.rs | 54 ------------------- .../src/runtime/request_timing_tests.rs | 3 +- 6 files changed, 21 insertions(+), 64 deletions(-) delete mode 100644 moli-fetch/src/runtime/request_timing.rs 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/cancellation.rs b/moli-fetch/src/cancellation.rs index cb975c29bbe..bd454db69be 100644 --- a/moli-fetch/src/cancellation.rs +++ b/moli-fetch/src/cancellation.rs @@ -112,7 +112,7 @@ impl FetchResponseTiming { self.body_bytes_received } /// The transport's pre-request boundary, after connection/TLS setup and - /// before sending the request. Absent for responses served from cache. + /// before sending the request. Pure cache hits do not produce this boundary. pub fn request_started_at(self) -> Option { self.request_started_at } diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index 223b7ad6b31..86456b33b36 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -61,8 +61,6 @@ use crate::{ proxy_connect::{ProxyConnectResponse, ProxyConnectResponseRecorder}, }; -mod request_timing; - const DEFAULT_RUNTIME_TRANSFERS: usize = 256; const RUNTIME_POLL_INTERVAL: Duration = Duration::from_millis(50); // curl-sys does not currently expose CURLINFO_HTTP_VERSION. This value is @@ -2842,6 +2840,19 @@ 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(); @@ -2914,7 +2925,8 @@ fn configure_network_observation( } easy.get_mut() .begin_transfer(recorder, capture_proxy_connect_response); - request_timing::configure(easy)?; + easy.prereq(true) + .context("failed to configure curl pre-request callback")?; easy.verbose(verbose) .context("failed to configure curl network observation") } diff --git a/moli-fetch/src/runtime/request_timing.rs b/moli-fetch/src/runtime/request_timing.rs deleted file mode 100644 index 358d17110b1..00000000000 --- a/moli-fetch/src/runtime/request_timing.rs +++ /dev/null @@ -1,54 +0,0 @@ -use std::{ - ffi::{c_char, c_int, c_void}, - panic::{AssertUnwindSafe, catch_unwind}, -}; - -use super::*; - -// curl-sys does not expose these options yet. Both values are part of curl's -// public curl.h ABI, available since libcurl 7.80.0. -const CURLOPT_PREREQFUNCTION: curl_sys::CURLoption = curl_sys::CURLOPTTYPE_FUNCTIONPOINT + 312; -const CURLOPT_PREREQDATA: curl_sys::CURLoption = curl_sys::CURLOPTTYPE_OBJECTPOINT + 313; - -pub(super) fn configure(easy: &mut Easy2) -> Result<()> { - let callback: extern "C" fn(*mut c_void, *mut c_char, *mut c_char, c_int, c_int) -> c_int = - request_started; - let handler = std::ptr::from_mut(easy.get_mut()).cast::(); - // SAFETY: curl-rust keeps the handler in Easy2's boxed Inner allocation. - // Moving Easy2 does not move the handler, and curl_easy_cleanup runs before - // that allocation is freed. libcurl calls this synchronously on the same - // thread as the other Handler callbacks, with no overlapping handler access. - // Every attempt reconfigures these options after Easy2::reset; its collector - // is bound to that attempt's ResponseProgress before the easy is submitted. - let code = unsafe { curl_sys::curl_easy_setopt(easy.raw(), CURLOPT_PREREQDATA, handler) }; - if code != curl_sys::CURLE_OK { - return Err(curl::Error::new(code)).context("failed to configure curl pre-request data"); - } - // SAFETY: callback has the curl_prereq_callback signature from curl.h. - let code = unsafe { curl_sys::curl_easy_setopt(easy.raw(), CURLOPT_PREREQFUNCTION, callback) }; - if code != curl_sys::CURLE_OK { - return Err(curl::Error::new(code)) - .context("failed to configure curl pre-request callback"); - } - Ok(()) -} - -extern "C" fn request_started( - handler: *mut c_void, - _remote_ip: *mut c_char, - _local_ip: *mut c_char, - _remote_port: c_int, - _local_port: c_int, -) -> c_int { - // Do not unwind across the C ABI. Abort this transfer if the callback fails. - catch_unwind(AssertUnwindSafe(|| { - // SAFETY: configure installs a live FetchTransferHandler as PREREQDATA; - // libcurl invokes callbacks serially while its Easy2 remains alive. - let handler = unsafe { &*handler.cast::() }; - if let Some(timing) = handler.response_progress() { - timing.mark_request_started(); - } - 0 // CURL_PREREQFUNC_OK - })) - .unwrap_or(1) // CURL_PREREQFUNC_ABORT -} diff --git a/moli-fetch/src/runtime/request_timing_tests.rs b/moli-fetch/src/runtime/request_timing_tests.rs index e17def320e5..6532fde702f 100644 --- a/moli-fetch/src/runtime/request_timing_tests.rs +++ b/moli-fetch/src/runtime/request_timing_tests.rs @@ -49,8 +49,7 @@ fn read_head(stream: &mut std::net::TcpStream) -> std::io::Result> { } #[test] -fn prerequest_timing_survives_reuse_reset_moves_and_progress_reselection_without_verbose() --> Result<()> { +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<()> {