fix(fetch): align HTML transfer timing and reduce observer overhead

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" }

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<usize>,
     response_too_large: bool,
     response_bytes_received: usize,
+    declared_identity_body_length: Option<NonZeroUsize>,
     cache_body_writer: Option<HttpCacheBodyWriter>,
     cache_plan: Option<StreamingCachePlan>,
     started: bool,
@@ -257,7 +258,7 @@ pub struct RawStreamingResponseCollector {
     max_response_size: Option<usize>,
     response_too_large: bool,
     response_bytes_received: usize,
-    declared_identity_body_length: Option<usize>,
+    declared_identity_body_length: Option<NonZeroUsize>,
     cache_body_writer: Option<HttpCacheBodyWriter>,
     cache_plan: Option<StreamingCachePlan>,
     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<RedirectInfo>,
         network_request_extra_info: Option<NetworkRequestExtraInfo>,
         cache_plan: Option<StreamingCachePlan>,
+        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();

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<AtomicU8>,
-    timing: Arc<Mutex<FetchResponseTiming>>,
+    inner: Arc<ResponseProgressState>,
+}
+
+#[derive(Debug, Default)]
+struct ResponseProgressState {
+    state: AtomicU8,
+    timing: Mutex<FetchResponseTiming>,
 }
 
 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<Instant> {
         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)]