Coalesce cacheable resource fetches

crates/obscura-browser/src/page.rs

@@ -1355,11 +1355,16 @@ impl Page {
 
         let client = self.http_client.clone();
         let page_callbacks = self.callbacks.clone();
+        let script_initiator = self
+            .url
+            .clone()
+            .unwrap_or_else(|| Url::parse("about:blank").unwrap());
         let fetch_futures: Vec<_> = fetch_tasks
             .iter()
             .map(|(idx, url)| {
                 let client = client.clone();
                 let cbs = page_callbacks.clone();
+                let initiator = script_initiator.clone();
                 let url = url.clone();
                 let idx = *idx;
                 async move {
@@ -1389,7 +1394,11 @@ impl Page {
                         };
                         return Some((idx, url, resp));
                     }
-                    match client.fetch_with_callbacks(&parsed, Some(&cbs)).await {
+                    let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+                    match client
+                        .fetch_resource_with_callbacks(&parsed, request, Some(&cbs))
+                        .await
+                    {
                         Ok(resp) => Some((idx, url, resp)),
                         Err(e) => {
                             tracing::warn!("Failed to fetch script {}: {}", url, e);
@@ -3764,6 +3773,137 @@ mod tests {
         (format!("http://{address}"), request_rx)
     }
 
+    fn spawn_script_resource_cache_server(
+        distinct: bool,
+    ) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
+        use std::io::{Read as _, Write as _};
+        use std::sync::atomic::Ordering;
+
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let address = listener.local_addr().unwrap();
+        let script_requests = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let observed_requests = script_requests.clone();
+        std::thread::spawn(move || {
+            for stream in listener.incoming() {
+                let Ok(mut stream) = stream else {
+                    continue;
+                };
+                let observed_requests = observed_requests.clone();
+                std::thread::spawn(move || {
+                    let mut request = [0u8; 2048];
+                    let length = stream.read(&mut request).unwrap_or(0);
+                    let request_text = String::from_utf8_lossy(&request[..length]);
+                    let path = request_text
+                        .lines()
+                        .next()
+                        .and_then(|line| line.split_ascii_whitespace().nth(1))
+                        .unwrap_or("/");
+                    let (content_type, cache_control, body) = if path == "/duplicate.html" {
+                        let tags = (0..32)
+                            .map(|_| "<script src='/shared.js'></script>")
+                            .collect::<String>();
+                        (
+                            "text/html",
+                            "no-store",
+                            format!(
+                                "<!doctype html><html><body><script>globalThis.__runs=0</script>{tags}</body></html>"
+                            ),
+                        )
+                    } else if path == "/distinct.html" {
+                        let tags = (0..24)
+                            .map(|index| format!("<script src='/distinct/{index}.js'></script>"))
+                            .collect::<String>();
+                        (
+                            "text/html",
+                            "no-store",
+                            format!(
+                                "<!doctype html><html><body><script>globalThis.__runs=0</script>{tags}</body></html>"
+                            ),
+                        )
+                    } else if path == "/shared.js" || path.starts_with("/distinct/") {
+                        observed_requests.fetch_add(1, Ordering::SeqCst);
+                        std::thread::sleep(std::time::Duration::from_millis(80));
+                        (
+                            "application/javascript",
+                            "public, max-age=3600",
+                            "globalThis.__runs=(globalThis.__runs||0)+1;".to_string(),
+                        )
+                    } else {
+                        ("text/plain", "no-store", "not found".to_string())
+                    };
+                    let response = format!(
+                        "HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nCache-Control: {cache_control}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+                        body.len(),
+                    );
+                    let _ = stream.write_all(response.as_bytes());
+                });
+            }
+        });
+        let page = if distinct {
+            "distinct.html"
+        } else {
+            "duplicate.html"
+        };
+        (format!("http://{address}/{page}"), script_requests)
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn duplicate_cacheable_scripts_fetch_once_but_execute_for_each_element() {
+        use std::sync::atomic::Ordering;
+
+        let (url, script_requests) = spawn_script_resource_cache_server(false);
+        let context = std::sync::Arc::new(crate::BrowserContext::with_storage_and_network(
+            "duplicate-script-cache".to_string(),
+            None,
+            false,
+            None,
+            None,
+            true,
+        ));
+        let mut page = super::Page::new("duplicate-script-cache".to_string(), context);
+
+        page.navigate(&url).await.unwrap();
+
+        assert_eq!(
+            page.js
+                .as_mut()
+                .unwrap()
+                .evaluate("globalThis.__runs")
+                .unwrap(),
+            serde_json::json!(32.0),
+            "a cached response must still execute for every script element",
+        );
+        assert_eq!(script_requests.load(Ordering::SeqCst), 1);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn distinct_cacheable_scripts_keep_distinct_network_requests() {
+        use std::sync::atomic::Ordering;
+
+        let (url, script_requests) = spawn_script_resource_cache_server(true);
+        let context = std::sync::Arc::new(crate::BrowserContext::with_storage_and_network(
+            "distinct-script-cache".to_string(),
+            None,
+            false,
+            None,
+            None,
+            true,
+        ));
+        let mut page = super::Page::new("distinct-script-cache".to_string(), context);
+
+        page.navigate(&url).await.unwrap();
+
+        assert_eq!(
+            page.js
+                .as_mut()
+                .unwrap()
+                .evaluate("globalThis.__runs")
+                .unwrap(),
+            serde_json::json!(24.0),
+        );
+        assert_eq!(script_requests.load(Ordering::SeqCst), 24);
+    }
+
     #[test]
     fn external_scripts_require_a_successful_http_status() {
         assert!(script_response_is_executable(200));

crates/obscura-net/src/client.rs

@@ -1,7 +1,7 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::sync::{Arc, OnceLock};
 use std::sync::atomic::{AtomicU32, Ordering};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use std::net::{IpAddr, SocketAddr};
 
@@ -9,7 +9,7 @@ use reqwest::dns::{Addrs, Name, Resolve, Resolving};
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue, USER_AGENT};
 use reqwest::redirect::Policy;
 use reqwest::{Client, Method};
-use tokio::sync::RwLock;
+use tokio::sync::{RwLock, watch};
 use url::Url;
 
 use crate::cookies::CookieJar;
@@ -124,7 +124,7 @@ pub struct RequestInfo {
     pub resource_type: ResourceType,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub enum ResourceType {
     Document,
     Script,
@@ -139,7 +139,7 @@ pub enum ResourceType {
 /// Fetch metadata for a browser-owned request. Navigation keeps its existing
 /// profile; render resources use this type so they do not masquerade as HTML
 /// documents when they move onto the page's asynchronous transport.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub enum RequestMode {
     Navigate,
     NoCors,
@@ -147,7 +147,7 @@ pub enum RequestMode {
     SameOrigin,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub enum RequestCredentials {
     Omit,
     SameOrigin,
@@ -813,12 +813,148 @@ pub struct ObscuraHttpClient {
     pub timeout: Duration,
     pub in_flight: Arc<std::sync::atomic::AtomicU32>,
     pub block_trackers: bool,
+    resource_loader: std::sync::Mutex<ResourceLoaderState>,
     /// When true, `validate_url` lets localhost / RFC1918 / link-local addresses
     /// through in addition to the `OBSCURA_ALLOW_PRIVATE_NETWORK` env var.
     /// Set via `--allow-private-network` on the CLI (issue #33).
     pub allow_private_network: bool,
 }
 
+const RESOURCE_CACHE_MAX_ENTRIES: usize = 256;
+const RESOURCE_CACHE_MAX_BYTES: usize = 64 * 1024 * 1024;
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+struct ResourceCacheKey {
+    url: String,
+    resource_type: ResourceType,
+    mode: RequestMode,
+    credentials: RequestCredentials,
+    initiator: Option<String>,
+    user_agent: String,
+    extra_headers: Vec<(String, String)>,
+    max_response_bytes: usize,
+}
+
+#[derive(Clone)]
+struct ResourceCacheEntry {
+    response: Response,
+    expires_at: Instant,
+}
+
+#[derive(Default)]
+struct ResourceCache {
+    entries: HashMap<ResourceCacheKey, ResourceCacheEntry>,
+    insertion_order: VecDeque<ResourceCacheKey>,
+    body_bytes: usize,
+}
+
+#[derive(Default)]
+struct ResourceLoaderState {
+    cache: ResourceCache,
+    shared_fetches: HashMap<ResourceCacheKey, SharedFetchSender>,
+}
+
+#[derive(Clone)]
+enum SharedFetchOutcome {
+    Cacheable(Response),
+    RetryUncoalesced,
+}
+
+type SharedFetchSender = watch::Sender<Option<SharedFetchOutcome>>;
+
+struct SharedFetchLeader<'a> {
+    loader: &'a std::sync::Mutex<ResourceLoaderState>,
+    key: ResourceCacheKey,
+    sender: SharedFetchSender,
+    finished: bool,
+}
+
+impl SharedFetchLeader<'_> {
+    fn finish(mut self, outcome: SharedFetchOutcome) {
+        self.loader.lock().unwrap().shared_fetches.remove(&self.key);
+        let _ = self.sender.send(Some(outcome));
+        self.finished = true;
+    }
+}
+
+impl Drop for SharedFetchLeader<'_> {
+    fn drop(&mut self) {
+        if self.finished {
+            return;
+        }
+        self.loader.lock().unwrap().shared_fetches.remove(&self.key);
+        let _ = self.sender.send(Some(SharedFetchOutcome::RetryUncoalesced));
+    }
+}
+
+impl ResourceCache {
+    fn get(&mut self, key: &ResourceCacheKey) -> Option<Response> {
+        let entry = self.entries.get(key)?;
+        if entry.expires_at <= Instant::now() {
+            let expired = self.entries.remove(key)?;
+            self.body_bytes = self.body_bytes.saturating_sub(expired.response.body.len());
+            return None;
+        }
+        Some(entry.response.clone())
+    }
+
+    fn insert(&mut self, key: ResourceCacheKey, response: Response, lifetime: Duration) {
+        let response_bytes = response.body.len();
+        if response_bytes > RESOURCE_CACHE_MAX_BYTES {
+            return;
+        }
+        if let Some(previous) = self.entries.remove(&key) {
+            self.body_bytes = self.body_bytes.saturating_sub(previous.response.body.len());
+            self.insertion_order.retain(|queued| queued != &key);
+        }
+        while self.entries.len() >= RESOURCE_CACHE_MAX_ENTRIES
+            || self.body_bytes.saturating_add(response_bytes) > RESOURCE_CACHE_MAX_BYTES
+        {
+            let Some(oldest) = self.insertion_order.pop_front() else {
+                break;
+            };
+            if let Some(entry) = self.entries.remove(&oldest) {
+                self.body_bytes = self.body_bytes.saturating_sub(entry.response.body.len());
+            }
+        }
+        self.body_bytes = self.body_bytes.saturating_add(response_bytes);
+        self.insertion_order.push_back(key.clone());
+        self.entries.insert(
+            key,
+            ResourceCacheEntry {
+                response,
+                expires_at: Instant::now() + lifetime,
+            },
+        );
+    }
+}
+
+fn response_cache_lifetime(response: &Response) -> Option<Duration> {
+    if !(200..300).contains(&response.status)
+        || !response.redirected_from.is_empty()
+        || response.header("set-cookie").is_some()
+        || response
+            .header("vary")
+            .is_some_and(|vary| vary.split(',').any(|name| name.trim() == "*"))
+    {
+        return None;
+    }
+    let cache_control = response.header("cache-control")?;
+    let mut max_age = None;
+    for directive in cache_control.split(',').map(str::trim) {
+        let lower = directive.to_ascii_lowercase();
+        if lower == "no-store" || lower == "no-cache" {
+            return None;
+        }
+        if let Some(value) = lower.strip_prefix("max-age=") {
+            max_age = value.trim_matches('"').parse::<u64>().ok();
+        }
+    }
+    max_age
+        .filter(|seconds| *seconds > 0)
+        .map(Duration::from_secs)
+}
+
 /// Derive the sec-ch-ua and sec-ch-ua-platform client-hint header values from a
 /// User-Agent string, using Chromium's per-major-version GREASE algorithm so
 /// the non-stealth HTTP path agrees with navigator.userAgentData instead of
@@ -889,6 +1025,7 @@ impl ObscuraHttpClient {
             in_flight: Arc::new(std::sync::atomic::AtomicU32::new(0)),
             timeout: Duration::from_secs(30),
             block_trackers: false,
+            resource_loader: std::sync::Mutex::new(ResourceLoaderState::default()),
             allow_private_network,
         }
     }
@@ -999,6 +1136,53 @@ impl ObscuraHttpClient {
             .await
     }
 
+    async fn resource_cache_key(
+        &self,
+        method: &Method,
+        url: &Url,
+        body: &Option<Vec<u8>>,
+        request: &ResourceRequest,
+    ) -> Option<ResourceCacheKey> {
+        if *method != Method::GET
+            || body.is_some()
+            || request.resource_type == ResourceType::Document
+            || !matches!(url.scheme(), "http" | "https")
+            || self.interceptor.read().await.is_some()
+        {
+            return None;
+        }
+        let mut extra_headers = self
+            .extra_headers
+            .read()
+            .await
+            .iter()
+            .map(|(name, value)| (name.to_ascii_lowercase(), value.clone()))
+            .collect::<Vec<_>>();
+        extra_headers.sort();
+        if extra_headers.iter().any(|(name, value)| {
+            name == "authorization"
+                || name == "cookie"
+                || (name == "cache-control"
+                    && (value.to_ascii_lowercase().contains("no-cache")
+                        || value.to_ascii_lowercase().contains("no-store")))
+        }) {
+            return None;
+        }
+        if request.sends_credentials_to(url) && !self.cookie_jar.get_cookie_header(url).is_empty() {
+            return None;
+        }
+        Some(ResourceCacheKey {
+            url: url.to_string(),
+            resource_type: request.resource_type,
+            mode: request.mode,
+            credentials: request.credentials,
+            initiator: request.initiator.as_ref().map(ToString::to_string),
+            user_agent: self.user_agent.read().await.clone(),
+            extra_headers,
+            max_response_bytes: request.max_response_bytes,
+        })
+    }
+
     async fn fetch_with_profile(
         &self,
         initial_method: Method,
@@ -1006,6 +1190,150 @@ impl ObscuraHttpClient {
         initial_body: Option<Vec<u8>>,
         callbacks: Option<&CallbackRegistry>,
         request: ResourceRequest,
+    ) -> Result<Response, ObscuraNetError> {
+        let Some(cache_key) = self
+            .resource_cache_key(&initial_method, url, &initial_body, &request)
+            .await
+        else {
+            return self
+                .fetch_with_profile_uncached(initial_method, url, initial_body, callbacks, request)
+                .await;
+        };
+
+        enum Acquisition {
+            Cached(Response),
+            Follower(watch::Receiver<Option<SharedFetchOutcome>>),
+            Leader(SharedFetchSender),
+        }
+
+        // Cache lookup and leader election are one critical section.  Without
+        // that atomicity a request can miss the cache, pause, and install a
+        // second leader after the first request has already populated it.
+        let acquisition = {
+            let mut loader = self.resource_loader.lock().unwrap();
+            if let Some(response) = loader.cache.get(&cache_key) {
+                Acquisition::Cached(response)
+            } else if let Some(sender) = loader.shared_fetches.get(&cache_key) {
+                Acquisition::Follower(sender.subscribe())
+            } else {
+                let (sender, _receiver) = watch::channel(None);
+                loader
+                    .shared_fetches
+                    .insert(cache_key.clone(), sender.clone());
+                Acquisition::Leader(sender)
+            }
+        };
+
+        match acquisition {
+            Acquisition::Cached(response) => {
+                self.fire_logical_resource_callbacks(callbacks, url, &request, &response)
+                    .await;
+                Ok(response)
+            }
+            Acquisition::Follower(mut receiver) => loop {
+                let outcome = { receiver.borrow().clone() };
+                if let Some(outcome) = outcome {
+                    break match outcome {
+                        SharedFetchOutcome::Cacheable(response) => {
+                            self.fire_logical_resource_callbacks(
+                                callbacks, url, &request, &response,
+                            )
+                            .await;
+                            Ok(response)
+                        }
+                        SharedFetchOutcome::RetryUncoalesced => {
+                            self.fetch_with_profile_uncached(
+                                initial_method,
+                                url,
+                                initial_body,
+                                callbacks,
+                                request,
+                            )
+                            .await
+                        }
+                    };
+                }
+                if receiver.changed().await.is_err() {
+                    break self
+                        .fetch_with_profile_uncached(
+                            initial_method,
+                            url,
+                            initial_body,
+                            callbacks,
+                            request,
+                        )
+                        .await;
+                }
+            },
+            Acquisition::Leader(sender) => {
+                // If this future is cancelled or panics while awaiting I/O,
+                // the guard removes the stale leader and wakes followers so
+                // they can retry instead of waiting forever.
+                let leader = SharedFetchLeader {
+                    loader: &self.resource_loader,
+                    key: cache_key.clone(),
+                    sender,
+                    finished: false,
+                };
+                let result = self
+                    .fetch_with_profile_uncached(
+                        initial_method,
+                        url,
+                        initial_body,
+                        callbacks,
+                        request,
+                    )
+                    .await;
+                let shared_outcome = match &result {
+                    Ok(response) => match response_cache_lifetime(response) {
+                        Some(lifetime) => {
+                            self.resource_loader.lock().unwrap().cache.insert(
+                                cache_key,
+                                response.clone(),
+                                lifetime,
+                            );
+                            SharedFetchOutcome::Cacheable(response.clone())
+                        }
+                        None => SharedFetchOutcome::RetryUncoalesced,
+                    },
+                    Err(_) => SharedFetchOutcome::RetryUncoalesced,
+                };
+                leader.finish(shared_outcome);
+                result
+            }
+        }
+    }
+
+    /// A cache hit or shared transport still represents an independent page
+    /// request.  Keep passive request/response observers at logical-resource
+    /// granularity even when only one HTTP transaction reaches the server.
+    async fn fire_logical_resource_callbacks(
+        &self,
+        callbacks: Option<&CallbackRegistry>,
+        url: &Url,
+        request: &ResourceRequest,
+        response: &Response,
+    ) {
+        let Some(callbacks) = callbacks else {
+            return;
+        };
+        let request_info = RequestInfo {
+            url: url.clone(),
+            method: Method::GET.to_string(),
+            headers: self.extra_headers.read().await.clone(),
+            resource_type: request.resource_type,
+        };
+        callbacks.fire_request(&request_info).await;
+        callbacks.fire_response(&request_info, response).await;
+    }
+
+    async fn fetch_with_profile_uncached(
+        &self,
+        initial_method: Method,
+        url: &Url,
+        initial_body: Option<Vec<u8>>,
+        callbacks: Option<&CallbackRegistry>,
+        request: ResourceRequest,
     ) -> Result<Response, ObscuraNetError> {
         validate_url(url, self.allow_private_network)?;
         validate_request_mode(&request, url)?;
@@ -1316,10 +1644,12 @@ mod ssrf_tests {
     };
     use crate::cookies::CookieJar;
     use reqwest::dns::{Name, Resolve};
+    use std::collections::HashMap;
     use std::net::IpAddr;
     use std::str::FromStr;
     use std::sync::Arc;
     use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Duration;
     use url::Url;
 
     fn ip(s: &str) -> IpAddr {
@@ -1612,6 +1942,52 @@ mod ssrf_tests {
         )
     }
 
+    async fn cancelled_shared_fetch_fixture(
+    ) -> (Url, tokio::sync::oneshot::Receiver<()>, Arc<AtomicUsize>) {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let requests = Arc::new(AtomicUsize::new(0));
+        let observed = requests.clone();
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        tokio::spawn(async move {
+            let mut first_stream = None;
+            let mut started_tx = Some(started_tx);
+            for index in 0..3 {
+                let Ok((mut stream, _)) = listener.accept().await else {
+                    return;
+                };
+                let mut request = [0u8; 2048];
+                let _ = stream.read(&mut request).await;
+                observed.fetch_add(1, Ordering::SeqCst);
+                if index == 0 {
+                    if let Some(started_tx) = started_tx.take() {
+                        let _ = started_tx.send(());
+                    }
+                    // Hold the transport open until the leader task is
+                    // cancelled. The next two connections prove both the
+                    // waiting follower retry and a fresh cache leader work.
+                    first_stream = Some(stream);
+                    continue;
+                }
+                let body = "shared";
+                let response = format!(
+                    "HTTP/1.1 200 OK\r\nCache-Control: public, max-age=3600\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+                    body.len(),
+                );
+                let _ = stream.write_all(response.as_bytes()).await;
+                let _ = stream.shutdown().await;
+            }
+            drop(first_stream);
+        });
+        (
+            Url::parse(&format!("http://{address}/shared.js")).unwrap(),
+            started_rx,
+            requests,
+        )
+    }
+
     #[tokio::test]
     async fn cancellation_returns_active_requests_to_zero() {
         let (target, started) = hanging_fixture().await;
@@ -1631,6 +2007,81 @@ mod ssrf_tests {
         assert_eq!(client.active_requests(), 0);
     }
 
+    #[tokio::test]
+    async fn cancelled_shared_subresource_leader_wakes_follower_and_clears_slot() {
+        let (target, started, network_requests) = cancelled_shared_fetch_fixture().await;
+        let initiator = target.join("/page.html").unwrap();
+        let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+        let client = Arc::new(ObscuraHttpClient::with_full_options(
+            Arc::new(CookieJar::new()),
+            None,
+            true,
+        ));
+
+        let leader = tokio::spawn({
+            let client = client.clone();
+            let target = target.clone();
+            let request = request.clone();
+            async move {
+                client
+                    .fetch_resource_with_callbacks(&target, request, None)
+                    .await
+            }
+        });
+        started.await.unwrap();
+
+        let follower = tokio::spawn({
+            let client = client.clone();
+            let target = target.clone();
+            let request = request.clone();
+            async move {
+                client
+                    .fetch_resource_with_callbacks(&target, request, None)
+                    .await
+            }
+        });
+        tokio::time::timeout(Duration::from_secs(2), async {
+            loop {
+                let follower_is_waiting = client
+                    .resource_loader
+                    .lock()
+                    .unwrap()
+                    .shared_fetches
+                    .values()
+                    .next()
+                    .is_some_and(|sender| sender.receiver_count() > 0);
+                if follower_is_waiting {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("follower did not join the shared fetch");
+
+        leader.abort();
+        let _ = leader.await;
+        let response = tokio::time::timeout(Duration::from_secs(2), follower)
+            .await
+            .expect("follower remained blocked after leader cancellation")
+            .unwrap()
+            .unwrap();
+        assert_eq!(response.body, b"shared");
+
+        // The follower intentionally retried without populating the cache.
+        // A subsequent request must be able to install a fresh leader, and
+        // its successful response is then reusable.
+        client
+            .fetch_resource_with_callbacks(&target, request.clone(), None)
+            .await
+            .unwrap();
+        client
+            .fetch_resource_with_callbacks(&target, request, None)
+            .await
+            .unwrap();
+        assert_eq!(network_requests.load(Ordering::SeqCst), 3);
+    }
+
     #[tokio::test]
     async fn transport_timeout_returns_active_requests_to_zero() {
         let (target, started) = hanging_fixture().await;
@@ -1675,6 +2126,184 @@ mod ssrf_tests {
         assert_eq!(responses.load(Ordering::SeqCst), 1);
     }
 
+    async fn cacheable_resource_fixture(
+        status: u16,
+        headers: &'static str,
+    ) -> (Url, Arc<AtomicUsize>) {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let requests = Arc::new(AtomicUsize::new(0));
+        let observed = requests.clone();
+        tokio::spawn(async move {
+            while let Ok((mut stream, _)) = listener.accept().await {
+                let observed = observed.clone();
+                tokio::spawn(async move {
+                    let mut request = [0u8; 2048];
+                    let _ = stream.read(&mut request).await;
+                    observed.fetch_add(1, Ordering::SeqCst);
+                    tokio::time::sleep(Duration::from_millis(80)).await;
+                    let body = "globalThis.__sharedRuns=(globalThis.__sharedRuns||0)+1;";
+                    let response = format!(
+                        "HTTP/1.1 {status} Test\r\nContent-Type: application/javascript\r\nContent-Length: {}\r\n{headers}Connection: close\r\n\r\n{body}",
+                        body.len(),
+                    );
+                    let _ = stream.write_all(response.as_bytes()).await;
+                    let _ = stream.shutdown().await;
+                });
+            }
+        });
+        (
+            Url::parse(&format!("http://{address}/shared.js")).unwrap(),
+            requests,
+        )
+    }
+
+    #[tokio::test]
+    async fn cacheable_identical_subresources_share_one_in_flight_request() {
+        let (url, network_requests) = cacheable_resource_fixture(
+            200,
+            "Cache-Control: public, max-age=3600\r\nVary: Accept-Language\r\n",
+        )
+        .await;
+        let initiator = url.join("/page.html").unwrap();
+        let client = Arc::new(ObscuraHttpClient::with_full_options(
+            Arc::new(CookieJar::new()),
+            None,
+            true,
+        ));
+        let callbacks = Arc::new(CallbackRegistry::new());
+        let callback_requests = Arc::new(AtomicUsize::new(0));
+        let callback_responses = Arc::new(AtomicUsize::new(0));
+        let observed_requests = callback_requests.clone();
+        callbacks.add_request(Arc::new(move |_| {
+            observed_requests.fetch_add(1, Ordering::SeqCst);
+        }));
+        let observed_responses = callback_responses.clone();
+        callbacks.add_response(Arc::new(move |_, _| {
+            observed_responses.fetch_add(1, Ordering::SeqCst);
+        }));
+
+        let mut fetches = tokio::task::JoinSet::new();
+        for _ in 0..32 {
+            let client = client.clone();
+            let callbacks = callbacks.clone();
+            let url = url.clone();
+            let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+            fetches.spawn(async move {
+                client
+                    .fetch_resource_with_callbacks(&url, request, Some(&callbacks))
+                    .await
+                    .unwrap()
+            });
+        }
+        let mut responses = Vec::new();
+        while let Some(response) = fetches.join_next().await {
+            responses.push(response.unwrap());
+        }
+
+        assert_eq!(responses.len(), 32);
+        assert!(responses.iter().all(|response| response.status == 200));
+        assert_eq!(network_requests.load(Ordering::SeqCst), 1);
+        assert_eq!(callback_requests.load(Ordering::SeqCst), 32);
+        assert_eq!(callback_responses.load(Ordering::SeqCst), 32);
+    }
+
+    #[tokio::test]
+    async fn distinct_subresource_urls_do_not_coalesce() {
+        let (url, network_requests) =
+            cacheable_resource_fixture(200, "Cache-Control: public, max-age=3600\r\n").await;
+        let initiator = url.join("/page.html").unwrap();
+        let client = Arc::new(ObscuraHttpClient::with_full_options(
+            Arc::new(CookieJar::new()),
+            None,
+            true,
+        ));
+
+        let mut fetches = tokio::task::JoinSet::new();
+        for index in 0..24 {
+            let client = client.clone();
+            let url = url.join(&format!("/distinct/{index}.js")).unwrap();
+            let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+            fetches.spawn(async move {
+                client
+                    .fetch_resource_with_callbacks(&url, request, None)
+                    .await
+                    .unwrap()
+            });
+        }
+        let mut responses = Vec::new();
+        while let Some(response) = fetches.join_next().await {
+            responses.push(response.unwrap());
+        }
+
+        assert_eq!(responses.len(), 24);
+        assert_eq!(network_requests.load(Ordering::SeqCst), 24);
+    }
+
+    #[tokio::test]
+    async fn no_store_vary_star_and_error_responses_are_not_reused() {
+        for (status, headers) in [
+            (200, "Cache-Control: no-store\r\n"),
+            (200, "Cache-Control: public, max-age=3600\r\nVary: *\r\n"),
+            (500, "Cache-Control: public, max-age=3600\r\n"),
+        ] {
+            let (url, network_requests) = cacheable_resource_fixture(status, headers).await;
+            let initiator = url.join("/page.html").unwrap();
+            let client =
+                ObscuraHttpClient::with_full_options(Arc::new(CookieJar::new()), None, true);
+            let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+            client
+                .fetch_resource_with_callbacks(&url, request.clone(), None)
+                .await
+                .unwrap();
+            client
+                .fetch_resource_with_callbacks(&url, request, None)
+                .await
+                .unwrap();
+            assert_eq!(
+                network_requests.load(Ordering::SeqCst),
+                2,
+                "status={status} headers={headers:?}",
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn authorization_and_cookie_bearing_requests_bypass_resource_cache() {
+        for header in [
+            ("Authorization", "Bearer secret"),
+            ("Cookie", "session=secret"),
+        ] {
+            let (url, network_requests) =
+                cacheable_resource_fixture(200, "Cache-Control: public, max-age=3600\r\n").await;
+            let initiator = url.join("/page.html").unwrap();
+            let client =
+                ObscuraHttpClient::with_full_options(Arc::new(CookieJar::new()), None, true);
+            client
+                .set_extra_headers(HashMap::from([(
+                    header.0.to_string(),
+                    header.1.to_string(),
+                )]))
+                .await;
+            let request = ResourceRequest::subresource(ResourceType::Script, &initiator);
+            client
+                .fetch_resource_with_callbacks(&url, request.clone(), None)
+                .await
+                .unwrap();
+            client
+                .fetch_resource_with_callbacks(&url, request, None)
+                .await
+                .unwrap();
+            assert_eq!(
+                network_requests.load(Ordering::SeqCst),
+                2,
+                "header={header:?}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn resolver_blocks_hostname_that_resolves_to_loopback() {
         // localtest.me is a public DNS name that resolves to 127.0.0.1 — the