From f18a7c7394ae4466d38c31ff98db35142d49d7d8 Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:36:01 +0100 Subject: [PATCH 1/3] test: spawn the proxy with several workers --- tests/integration/common.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/tests/integration/common.rs b/tests/integration/common.rs index 4493b36..927ed9d 100644 --- a/tests/integration/common.rs +++ b/tests/integration/common.rs @@ -95,10 +95,20 @@ pub fn spawn(origin_url: &str) -> Running { /// `spawn` with a customised config. pub fn spawn_with(origin_url: &str, customise: impl FnOnce(&mut Config)) -> Running { + spawn_workers(origin_url, 1, customise) +} + +/// `spawn_with` on `workers` worker threads, so that concurrent connections +/// can land on different workers. +pub fn spawn_workers( + origin_url: &str, + workers: usize, + customise: impl FnOnce(&mut Config), +) -> Running { let (state, assets) = state_with(origin_url, customise); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); - let server = run(state, listener, None, 1).unwrap(); + let server = run(state, listener, None, workers).unwrap(); let handle = server.handle(); actix_web::rt::spawn(server); Running { From 87cad0da40de743d5500437c5656367b955591c7 Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:36:17 +0100 Subject: [PATCH 2/3] feat(proxy): coalesce concurrent misses for a key --- src/cache.rs | 41 ++ src/coalesce.rs | 154 ++++++++ src/lib.rs | 20 +- src/proxy.rs | 159 +++++++- tests/integration/coalescing.rs | 656 ++++++++++++++++++++++++++++++++ tests/integration/main.rs | 1 + 6 files changed, 1017 insertions(+), 14 deletions(-) create mode 100644 src/coalesce.rs create mode 100644 tests/integration/coalescing.rs diff --git a/src/cache.rs b/src/cache.rs index 770c64b..a5a16c9 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -61,6 +61,15 @@ struct ResponseKey { vary: Vec>, } +/// the key that concurrent requests coalesce on: the primary key and the +/// request's values for the `Vary` field names the store knows for it, so +/// that requests for different known variants do not wait for each other. +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +pub struct FlightKey { + primary: PrimaryKey, + vary: Vec>, +} + #[derive(Clone, Debug, Hash, PartialEq, Eq)] enum Key { Asset(PathBuf), @@ -386,6 +395,23 @@ impl Store { Some(Lookup::Stale(StaleEntry { response, key })) } + /// the coalescing key for a request, or `None` when the store keeps + /// nothing, so that no request could be answered from a leader's + /// response. + pub fn flight_key(&self, primary: &PrimaryKey, request: &HeaderMap) -> Option { + if !self.enabled() { + return None; + } + let vary = self + .index + .get(primary) + .map_or_else(Vec::new, |index| vary_values(&index.vary, request)); + Some(FlightKey { + primary: primary.clone(), + vary, + }) + } + /// claims the background revalidation of `entry`, or `None` when one is /// already running. pub fn start_refresh(&self, entry: &StaleEntry) -> Option { @@ -673,6 +699,21 @@ impl RequestPolicy { pub fn may_serve_stale(&self) -> bool { self.max_age.is_none() } + + /// whether the request may wait for a concurrent request for the same + /// key and then be answered from the cache. a request with credentials + /// does not, because the origin may answer it for that user alone, and + /// neither does one with its own preconditions, whose answer is for + /// that client. + pub fn may_coalesce(&self) -> bool { + self.may_serve && !self.has_credentials && !self.conditional + } + + /// whether other requests may wait for this one's response to be + /// stored. only a GET's response is stored. + pub fn may_lead(&self) -> bool { + self.may_coalesce() && self.may_store + } } /// a response that the store may keep, and for how long. diff --git a/src/coalesce.rs b/src/coalesce.rs new file mode 100644 index 0000000..082eb43 --- /dev/null +++ b/src/coalesce.rs @@ -0,0 +1,154 @@ +//! request coalescing: while one request for a cache key is on its way to the +//! origin, other requests for the key wait for it and then read the cache. + +use parking_lot::Mutex; +use std::collections::HashMap; +use std::hash::Hash; +use std::sync::Arc; +use tokio::sync::watch; + +type InFlight = Arc>>>; + +/// the keys with a request in flight, shared by every worker. +pub struct Flights { + in_flight: InFlight, +} + +impl Default for Flights { + fn default() -> Self { + Flights { + in_flight: Arc::default(), + } + } +} + +/// a request's part in the flight for its key. +pub enum Join { + /// the request goes to the origin, and others wait until the guard drops + Lead(FlightGuard), + /// another request is in flight for the key + Follow(Waiter), + /// nothing is in flight and the request may not lead + Alone, +} + +impl Flights { + /// joins the flight for `key`, or starts one when there is none and + /// `may_lead` holds. + pub fn join(&self, key: K, may_lead: bool) -> Join { + let mut in_flight = self.in_flight.lock(); + if let Some(done) = in_flight.get(&key) { + return Join::Follow(Waiter(done.clone())); + } + if !may_lead { + return Join::Alone; + } + let (sender, done) = watch::channel(()); + in_flight.insert(key.clone(), done); + Join::Lead(FlightGuard { + in_flight: self.in_flight.clone(), + key, + _sender: sender, + }) + } +} + +/// the leader's claim on a key. dropping it ends the flight and wakes the +/// followers. +pub struct FlightGuard { + in_flight: InFlight, + key: K, + /// dropped after `drop` has removed the key, so a woken follower never + /// finds the finished flight + _sender: watch::Sender<()>, +} + +impl Drop for FlightGuard { + fn drop(&mut self) { + self.in_flight.lock().remove(&self.key); + } +} + +/// a follower's view of a flight. +pub struct Waiter(watch::Receiver<()>); + +impl Waiter { + /// returns once the leader's guard has dropped. + pub async fn wait(mut self) { + // nothing is ever sent, so `changed` returns only when the sender + // drops, or at once if it already has + while self.0.changed().await.is_ok() {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn lead(flights: &Flights<&'static str>, key: &'static str) -> FlightGuard<&'static str> { + match flights.join(key, true) { + Join::Lead(guard) => guard, + _ => panic!("{key} did not lead"), + } + } + + fn follow(flights: &Flights<&'static str>, key: &'static str) -> Waiter { + match flights.join(key, true) { + Join::Follow(waiter) => waiter, + _ => panic!("{key} did not follow"), + } + } + + async fn released(waiter: Waiter) -> bool { + tokio::time::timeout(Duration::from_millis(100), waiter.wait()) + .await + .is_ok() + } + + #[tokio::test] + async fn followers_wait_until_the_guard_drops() { + let flights = Flights::default(); + let guard = lead(&flights, "a"); + + assert!(!released(follow(&flights, "a")).await); + let waiter = follow(&flights, "a"); + drop(guard); + assert!(released(waiter).await); + } + + #[tokio::test] + async fn a_follower_that_waits_after_the_guard_dropped_is_released() { + let flights = Flights::default(); + let guard = lead(&flights, "a"); + let waiter = follow(&flights, "a"); + drop(guard); + + assert!(released(waiter).await); + } + + #[test] + fn the_next_request_leads_once_the_flight_ends() { + let flights = Flights::default(); + drop(lead(&flights, "a")); + + lead(&flights, "a"); + } + + #[test] + fn keys_have_separate_flights() { + let flights = Flights::default(); + let _a = lead(&flights, "a"); + + lead(&flights, "b"); + } + + #[test] + fn a_request_that_may_not_lead_goes_alone() { + let flights = Flights::default(); + + assert!(matches!(flights.join("a", false), Join::Alone)); + let _guard = lead(&flights, "a"); + assert!(matches!(flights.join("a", false), Join::Follow(_))); + } +} diff --git a/src/lib.rs b/src/lib.rs index 5c6814e..0950741 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod tls; mod assets; mod cache; +mod coalesce; mod forwarded; mod proxy; @@ -32,12 +33,13 @@ use std::sync::Arc; use std::time::Duration; use url::Url; -use crate::cache::Store; +use crate::cache::{FlightKey, Store}; +use crate::coalesce::Flights; use crate::config::Config; /// how the cache answered requests. each proxied or asset response counts -/// once, as a hit, a miss, a revalidation or a stale serve. background -/// refreshes count the background revalidations started. +/// once, as a hit, a miss, a revalidation, a stale serve or a coalesced +/// serve. background refreshes count the background revalidations started. #[derive(Default)] pub struct CacheStats { hits: AtomicU64, @@ -45,6 +47,7 @@ pub struct CacheStats { revalidations: AtomicU64, stale: AtomicU64, background_refreshes: AtomicU64, + coalesced: AtomicU64, } impl CacheStats { @@ -69,12 +72,18 @@ impl CacheStats { fn background_refresh(&self) { self.background_refreshes.fetch_add(1, Ordering::Relaxed); } + + /// a response stored by a concurrent request that this one waited for + fn coalesced(&self) { + self.coalesced.fetch_add(1, Ordering::Relaxed); + } } /// application state, including cache pub struct AppState { cache_stats: CacheStats, cache: Store, + flights: Flights, http_client: Client, proxy::UpstreamBody>, upstream_base_url: Url, asset_path: PathBuf, @@ -88,8 +97,9 @@ async fn health_check(state: web::Data) -> impl Responder { let misses = stats.misses.load(Ordering::Relaxed); let revalidations = stats.revalidations.load(Ordering::Relaxed); let stale = stats.stale.load(Ordering::Relaxed); + let coalesced = stats.coalesced.load(Ordering::Relaxed); // the share of responses whose body came from the cache - let from_cache = hits + revalidations + stale; + let from_cache = hits + revalidations + stale + coalesced; let total = from_cache + misses; HttpResponse::Ok().json(serde_json::json!({ "status": "ok", @@ -98,6 +108,7 @@ async fn health_check(state: web::Data) -> impl Responder { "misses": misses, "revalidations": revalidations, "stale": stale, + "coalesced": coalesced, "background_refreshes": stats.background_refreshes.load(Ordering::Relaxed), "items": state.cache.entry_count(), "bytes": state.cache.weighted_size(), @@ -142,6 +153,7 @@ pub fn build_state(config: &Config) -> io::Result> { config.cache_size_mb, Duration::from_secs(config.cache_ttl_seconds), ), + flights: Flights::default(), http_client, upstream_base_url, asset_path: config.asset_path.clone(), diff --git a/src/proxy.rs b/src/proxy.rs index 8f52de3..6881d65 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -17,11 +17,12 @@ use tokio::sync::mpsc; use url::{Position, Url}; use crate::cache::{ - self, Fields, Lookup, PrimaryKey, RequestPolicy, StaleEntry, StoredResponse, + self, Fields, FlightKey, Lookup, PrimaryKey, RequestPolicy, StaleEntry, StoredResponse, PRECONDITION_HEADERS, }; +use crate::coalesce::{FlightGuard, Join, Waiter}; use crate::forwarded::{ClientInfo, CLIENT_FORWARDING_HEADERS, URL_OVERRIDE_HEADERS}; -use crate::AppState; +use crate::{AppState, CacheStats}; const CACHE_STATUS: &str = "x-shadowstep-cache"; @@ -30,6 +31,8 @@ type BodyError = Box; /// the body of a request to the origin. pub(crate) type UpstreamBody = UnsyncBoxBody; +type Flight = FlightGuard; + pub async fn forward_to_upstream( req: HttpRequest, payload: web::Payload, @@ -49,11 +52,11 @@ pub async fn forward_to_upstream( // X-Forwarded-Proto and X-Forwarded-Host that the origin receives let cache_key = PrimaryKey::new(client.scheme, &client.host, path_and_query); let request_policy = RequestPolicy::new(req.method(), req.headers()); - let stale = match from_cache(&req, &state, &cache_key, &request_policy) { - Cached::Answer(response) => return response, - Cached::Stale(entry) => Some(entry), - Cached::Nothing => None, - }; + let ToOrigin { stale, flight } = + match serve_from_cache(&req, &state, &cache_key, &request_policy).await { + Before::Answer(response) => return response, + Before::Origin(to_origin) => to_origin, + }; let Some((builder, target_uri)) = upstream_request(&req, &state, &client, path_and_query) else { @@ -74,6 +77,7 @@ pub async fn forward_to_upstream( cache_key, request_policy, stale, + flight, }; let hyper_req = match origin_request(builder, payload).await { @@ -112,6 +116,107 @@ pub async fn forward_to_upstream( } } +/// what the cache did for a request before any origin request. +enum Before { + /// the cache answers the request + Answer(HttpResponse), + /// the request goes to the origin + Origin(ToOrigin), +} + +/// what a request takes to the origin. +struct ToOrigin { + /// the stored response to revalidate + stale: Option, + /// the flight this request leads, which other requests wait for + flight: Option, +} + +/// the cache's answer to the request, if it has one. a request that may +/// coalesce and finds no response it can use as is joins the flight for +/// its key: the first such request leads it to the origin, and the others +/// wait for the leader to finish and then look again. +async fn serve_from_cache( + req: &HttpRequest, + state: &web::Data, + cache_key: &PrimaryKey, + request_policy: &RequestPolicy, +) -> Before { + let to_origin = match from_cache(req, state, cache_key, request_policy, FreshUse::Hit).or(None) + { + Before::Origin(to_origin) => to_origin, + answer => return answer, + }; + let (flight, fresh_use) = match join_flight(req, state, cache_key, request_policy) { + Join::Alone => return Before::Origin(to_origin), + Join::Lead(flight) => (Some(flight), FreshUse::Hit), + Join::Follow(waiter) => { + wait_for_leader(req, state, waiter).await; + (None, FreshUse::Coalesced) + } + }; + // a leader looks again because a flight for the key may have stored a + // response and ended since the first lookup. a follower that still + // finds nothing it may use goes to the origin alone, because the + // leader's response was not storable, did not match this request's + // `Vary` values or failed, and RFC 9111 section 3 lets a shared cache + // pass on only what it stores. + from_cache(req, state, cache_key, request_policy, fresh_use).or(flight) +} + +/// the request's part in the flight for its key. +fn join_flight( + req: &HttpRequest, + state: &AppState, + cache_key: &PrimaryKey, + request_policy: &RequestPolicy, +) -> Join { + if !request_policy.may_coalesce() { + return Join::Alone; + } + match state.cache.flight_key(cache_key, req.headers()) { + Some(key) => state.flights.join(key, request_policy.may_lead()), + None => Join::Alone, + } +} + +/// waits for the leader of a flight for no longer than the leader may wait +/// for the origin's response headers. a follower whose wait runs out goes +/// to the origin itself. +async fn wait_for_leader(req: &HttpRequest, state: &AppState, waiter: Waiter) { + if tokio::time::timeout(state.upstream_timeout, waiter.wait()) + .await + .is_err() + { + debug!("Stopped waiting for a concurrent request for {}", req.uri()); + } +} + +/// how a fresh stored response came to answer a request. +#[derive(Clone, Copy)] +enum FreshUse { + /// found on the request's own lookup + Hit, + /// stored by the flight this request waited for + Coalesced, +} + +impl FreshUse { + /// counts the use and returns its `X-Shadowstep-Cache` value. + fn count(self, stats: &CacheStats) -> &'static str { + match self { + FreshUse::Hit => { + stats.hit(); + "HIT" + } + FreshUse::Coalesced => { + stats.coalesced(); + "COALESCED" + } + } + } +} + /// what the cache can do for a request before it goes to the origin. enum Cached { /// the cache answers the request @@ -122,6 +227,18 @@ enum Cached { Nothing, } +impl Cached { + /// the cache's answer, or the origin request with `flight`. + fn or(self, flight: Option) -> Before { + let stale = match self { + Cached::Answer(response) => return Before::Answer(response), + Cached::Stale(entry) => Some(entry), + Cached::Nothing => None, + }; + Before::Origin(ToOrigin { stale, flight }) + } +} + /// the stored response for this request, if any, and whether it answers the /// request. only GET requests use stale responses, because a revalidation /// or a stored replacement needs a GET to the origin. @@ -130,6 +247,7 @@ fn from_cache( state: &web::Data, cache_key: &PrimaryKey, request_policy: &RequestPolicy, + fresh_use: FreshUse, ) -> Cached { if !request_policy.may_serve { return Cached::Nothing; @@ -140,8 +258,8 @@ fn from_cache( { Some(Lookup::Fresh(stored)) => { debug!("Cache hit for {} {}", req.method(), req.uri()); - state.cache_stats.hit(); - Cached::Answer(cached_response(&stored, "HIT")) + let cache_status = fresh_use.count(&state.cache_stats); + Cached::Answer(cached_response(&stored, cache_status)) } Some(Lookup::Stale(entry)) if req.method() == Method::GET => { while_revalidating(req, state, cache_key, request_policy, entry) @@ -268,6 +386,9 @@ struct Exchange { stale: Option, /// whether the origin request carries the stale response's validators revalidating: bool, + /// the flight this request leads. it ends when the exchange does, or + /// once the response is stored or known not to be + flight: Option, } impl Exchange { @@ -284,6 +405,7 @@ impl Exchange { } } self.state.cache_stats.miss(); + let flight = self.flight; let store = store_plan( &self.req, &self.state, @@ -291,7 +413,7 @@ impl Exchange { &self.request_policy, &head, ); - client_response(head, body, store) + client_response(head, body, store.map(|store| store.holding(flight))) } /// the client's response when the origin could not be reached or sent no @@ -547,6 +669,23 @@ struct Storing { finish: Box, } +impl Storing { + /// keeps `flight` until the response is stored, or until the copy of + /// its body is given up because it passed the limit, failed or was + /// dropped with the client's response, so that followers look up the + /// cache only once the outcome is known. + fn holding(self, flight: Option) -> Self { + let Storing { limit, finish } = self; + Storing { + limit, + finish: Box::new(move |status, headers, body| { + finish(status, headers, body); + drop(flight); + }), + } + } +} + /// a body on its way into the cache. struct BodyCopy { buffer: BytesMut, diff --git a/tests/integration/coalescing.rs b/tests/integration/coalescing.rs new file mode 100644 index 0000000..fe59176 --- /dev/null +++ b/tests/integration/coalescing.rs @@ -0,0 +1,656 @@ +use crate::common::{self, Running}; + +use bytes::Bytes; +use futures_util::future::join_all; +use http_body_util::Empty; +use hyper_util::client::legacy::connect::HttpConnector; +use hyper_util::client::legacy::Client; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::timeout; +use wiremock::matchers::{any, header, path}; +use wiremock::{Mock, MockServer, Request, ResponseTemplate}; + +const KIB: usize = 1024; +const MIB: usize = 1024 * KIB; + +/// how long the origin takes to answer, which is the window in which +/// followers join the leader. +const ORIGIN_DELAY: Duration = Duration::from_secs(1); + +/// longer than any step in these tests should take on a loaded CI runner, +/// and shorter than the default upstream timeout of 30 seconds, so that a +/// follower left waiting for its whole bounded wait fails the test. +const PATIENCE: Duration = Duration::from_secs(10); + +/// how many followers each test sends after its leader. +const FOLLOWERS: usize = 5; + +const WORKERS: usize = 4; + +type HttpClient = Client>; + +/// what a client got back from the proxy. +#[derive(Debug)] +struct Answer { + status: u16, + cache: String, + variant: String, + body: Bytes, +} + +async fn fetch(client: &HttpClient, url: &str, headers: &[(&str, &str)]) -> Answer { + let mut request = hyper::Request::get(url); + for (name, value) in headers { + request = request.header(*name, *value); + } + let response = timeout( + PATIENCE, + client.request(request.body(Empty::new()).unwrap()), + ) + .await + .expect("no response within the test's patience") + .unwrap(); + let text = |name: &str| { + response + .headers() + .get(name) + .map(|v| v.to_str().unwrap().to_owned()) + .unwrap_or_default() + }; + let (status, cache, variant) = ( + response.status().as_u16(), + text("x-shadowstep-cache"), + text("x-variant"), + ); + let body = timeout(PATIENCE, common::body_bytes(response.into_body())) + .await + .expect("no whole body within the test's patience"); + Answer { + status, + cache, + variant, + body, + } +} + +/// `n` concurrent GETs of `path`, each with `headers`. +async fn fetch_many( + client: &HttpClient, + server: &Running, + path: &str, + headers: &[(&str, &str)], + n: usize, +) -> Vec { + let url = server.url(path); + join_all((0..n).map(|_| fetch(client, &url, headers))).await +} + +/// starts a GET of `path` and returns once the origin has received it, so +/// that requests sent next find it in flight. +fn lead( + client: &HttpClient, + server: &Running, + path: &str, + headers: &[(&'static str, &'static str)], +) -> actix_web::rt::task::JoinHandle { + let (client, url, headers) = (client.clone(), server.url(path), headers.to_vec()); + actix_web::rt::spawn(async move { fetch(&client, &url, &headers).await }) +} + +async fn origin_requests(origin: &MockServer) -> usize { + origin.received_requests().await.unwrap().len() +} + +/// waits until the origin has received `n` requests. +async fn origin_receives(origin: &MockServer, n: usize) { + let started = Instant::now(); + while origin_requests(origin).await < n { + assert!( + started.elapsed() < PATIENCE, + "the origin never received {n} requests" + ); + actix_web::rt::time::sleep(Duration::from_millis(10)).await; + } +} + +async fn health(client: &HttpClient, server: &Running) -> serde_json::Value { + let answer = fetch(client, &server.url("/health"), &[]).await; + serde_json::from_slice(&answer.body).unwrap() +} + +async fn origin_responding(template: ResponseTemplate) -> MockServer { + let origin = MockServer::start().await; + Mock::given(any()) + .respond_with(template) + .mount(&origin) + .await; + origin +} + +fn cacheable(body: &str) -> ResponseTemplate { + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .set_body_string(body) + .set_delay(ORIGIN_DELAY) +} + +fn cache_statuses(answers: &[Answer]) -> Vec<&str> { + answers.iter().map(|a| a.cache.as_str()).collect() +} + +#[actix_web::test] +async fn concurrent_misses_reach_the_origin_once() { + let origin = origin_responding(cacheable("shared")).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + + let leader = lead(&client, &server, "/page", &[]); + origin_receives(&origin, 1).await; + let followers = fetch_many(&client, &server, "/page", &[], FOLLOWERS).await; + let leader = leader.await.unwrap(); + + assert_eq!((leader.status, leader.cache.as_str()), (200, "MISS")); + assert_eq!(leader.body, "shared"); + for follower in &followers { + assert_eq!(follower.status, 200); + assert_eq!(follower.body, "shared"); + } + assert_eq!(cache_statuses(&followers), ["COALESCED"; FOLLOWERS]); + assert_eq!(origin_requests(&origin).await, 1); + let health = health(&client, &server).await; + assert_eq!(health["cache"]["coalesced"], FOLLOWERS); + assert_eq!(health["cache"]["misses"], 1); + server.stop().await; +} + +/// answers with `template` and a body that echoes the request's +/// `X-Client`, so that each client can tell its own response. +fn echoing(template: ResponseTemplate) -> impl Fn(&Request) -> ResponseTemplate { + move |request: &Request| { + let client = request.headers.get("x-client").unwrap().to_str().unwrap(); + template.clone().set_body_string(client.to_owned()) + } +} + +#[actix_web::test] +async fn uncacheable_responses_are_not_shared() { + let templates = [ + ResponseTemplate::new(200).insert_header("cache-control", "private, max-age=60"), + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .insert_header("set-cookie", "session=1"), + ]; + for template in templates { + let origin = MockServer::start().await; + Mock::given(any()) + .respond_with(echoing(template.set_delay(ORIGIN_DELAY))) + .mount(&origin) + .await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + + let leader = lead(&client, &server, "/page", &[("x-client", "leader")]); + origin_receives(&origin, 1).await; + let url = server.url("/page"); + let names: Vec = (0..FOLLOWERS).map(|i| format!("follower-{i}")).collect(); + let followers = join_all(names.iter().map(|name| { + let (client, url) = (&client, &url); + async move { fetch(client, url, &[("x-client", name.as_str())]).await } + })) + .await; + let leader = leader.await.unwrap(); + + assert_eq!(leader.body, "leader"); + for (name, follower) in names.iter().zip(&followers) { + assert_eq!( + follower.body, + name.as_str(), + "got another client's response" + ); + assert_eq!(follower.cache, "MISS"); + } + assert_eq!(origin_requests(&origin).await, 1 + FOLLOWERS); + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; + } +} + +/// an origin that answers with `Vary: Accept-Encoding` and names the +/// variant it served in `X-Variant`. +async fn varying_origin(encodings: &[&str]) -> MockServer { + let origin = MockServer::start().await; + for encoding in encodings { + Mock::given(header("accept-encoding", *encoding)) + .respond_with( + cacheable("body") + .insert_header("vary", "Accept-Encoding") + .insert_header("x-variant", *encoding), + ) + .mount(&origin) + .await; + } + origin +} + +#[actix_web::test] +async fn followers_with_other_vary_values_go_to_the_origin() { + let origin = varying_origin(&["gzip", "br"]).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + + let leader = lead(&client, &server, "/page", &[("accept-encoding", "gzip")]); + origin_receives(&origin, 1).await; + let url = server.url("/page"); + let (gzip, br) = futures_util::join!( + join_all((0..3).map(|_| fetch(&client, &url, &[("accept-encoding", "gzip")]))), + join_all((0..3).map(|_| fetch(&client, &url, &[("accept-encoding", "br")]))), + ); + let leader = leader.await.unwrap(); + + assert_eq!( + (leader.cache.as_str(), leader.variant.as_str()), + ("MISS", "gzip") + ); + for answer in &gzip { + assert_eq!( + (answer.cache.as_str(), answer.variant.as_str()), + ("COALESCED", "gzip") + ); + } + for answer in &br { + assert_eq!( + (answer.cache.as_str(), answer.variant.as_str()), + ("MISS", "br") + ); + } + assert_eq!(origin_requests(&origin).await, 1 + 3); + server.stop().await; +} + +#[actix_web::test] +async fn known_vary_variants_coalesce_separately() { + let origin = varying_origin(&["deflate", "gzip", "br"]).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + // a stored variant tells the cache that the URL varies on + // Accept-Encoding before the other variants are requested + fetch( + &client, + &server.url("/page"), + &[("accept-encoding", "deflate")], + ) + .await; + + let gzip_leader = lead(&client, &server, "/page", &[("accept-encoding", "gzip")]); + origin_receives(&origin, 2).await; + let br_leader = lead(&client, &server, "/page", &[("accept-encoding", "br")]); + origin_receives(&origin, 3).await; + let url = server.url("/page"); + let (gzip, br) = futures_util::join!( + join_all((0..3).map(|_| fetch(&client, &url, &[("accept-encoding", "gzip")]))), + join_all((0..3).map(|_| fetch(&client, &url, &[("accept-encoding", "br")]))), + ); + + assert_eq!(br_leader.await.unwrap().cache, "MISS", "br followed gzip"); + assert_eq!(gzip_leader.await.unwrap().cache, "MISS"); + for (answers, encoding) in [(&gzip, "gzip"), (&br, "br")] { + for answer in answers { + assert_eq!( + (answer.cache.as_str(), answer.variant.as_str()), + ("COALESCED", encoding) + ); + } + } + assert_eq!(origin_requests(&origin).await, 3); + server.stop().await; +} + +#[actix_web::test] +async fn response_over_the_size_cap_is_not_shared() { + let body: Vec = (0..2 * MIB).map(|i| (i % 251) as u8).collect(); + let origin = origin_responding( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .set_body_bytes(body.clone()) + .set_delay(ORIGIN_DELAY), + ) + .await; + // a 1 MiB cache caps entries at 1 MiB + let server = common::spawn_workers(&origin.uri(), WORKERS, |c| c.cache_size_mb = 1); + let client = common::client(); + + let leader = lead(&client, &server, "/large", &[]); + origin_receives(&origin, 1).await; + let followers = fetch_many(&client, &server, "/large", &[], FOLLOWERS).await; + let leader = leader.await.unwrap(); + + for answer in followers.iter().chain([&leader]) { + assert_eq!((answer.status, answer.cache.as_str()), (200, "MISS")); + assert!(answer.body == body, "client received a different body"); + } + assert_eq!(origin_requests(&origin).await, 1 + FOLLOWERS); + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; +} + +#[actix_web::test] +async fn followers_of_a_failed_leader_get_their_own_answer() { + let origin = origin_responding(ResponseTemplate::new(502).set_delay(ORIGIN_DELAY)).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + + let leader = lead(&client, &server, "/page", &[]); + origin_receives(&origin, 1).await; + let followers = fetch_many(&client, &server, "/page", &[], FOLLOWERS).await; + let leader = leader.await.unwrap(); + + for answer in followers.iter().chain([&leader]) { + assert_eq!((answer.status, answer.cache.as_str()), (502, "MISS")); + } + assert_eq!(origin_requests(&origin).await, 1 + FOLLOWERS); + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; +} + +#[actix_web::test] +async fn followers_of_a_timed_out_leader_are_answered_in_time() { + let origin = origin_responding(cacheable("late").set_delay(Duration::from_secs(5))).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |c| { + c.upstream_timeout_seconds = 1; + }); + let client = common::client(); + + let started = Instant::now(); + let leader = lead(&client, &server, "/page", &[]); + origin_receives(&origin, 1).await; + let followers = fetch_many(&client, &server, "/page", &[], FOLLOWERS).await; + let leader = leader.await.unwrap(); + + for answer in followers.iter().chain([&leader]) { + assert_eq!(answer.status, 504); + } + // the leader's timeout releases the followers, and each then waits at + // most one timeout of its own + assert!( + started.elapsed() < Duration::from_secs(4), + "took {:?}", + started.elapsed() + ); + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; +} + +/// how the origin sends the rest of the first response's body. +#[derive(Clone, Copy)] +enum FirstBody { + /// one chunk every 100 ms, so the whole body takes over 6 seconds + Trickle, + /// never, holding the connection open + Stall, +} + +const CHUNK: usize = 16 * KIB; +const CHUNKS: usize = 64; + +/// an origin whose cacheable response is "first" and then `CHUNKS` chunks of +/// `CHUNK` bytes. the first connection gets the head and "first", then the +/// rest as `first_body` says. every later connection gets the whole +/// response at once. the counter holds the number of connections. +async fn origin_slowing_the_first_body(first_body: FirstBody) -> (String, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let connections = Arc::new(AtomicUsize::new(0)); + let counter = connections.clone(); + actix_web::rt::spawn(async move { + loop { + let (stream, _) = listener.accept().await.unwrap(); + let first = counter.fetch_add(1, Ordering::SeqCst) == 0; + actix_web::rt::spawn(answer_slowly(stream, first.then_some(first_body))); + } + }); + (url, connections) +} + +fn slow_body() -> Vec { + let mut body = b"first".to_vec(); + body.resize(5 + CHUNKS * CHUNK, b'x'); + body +} + +async fn answer_slowly(mut stream: TcpStream, first_body: Option) { + let mut head = Vec::new(); + let mut buf = [0; 1024]; + while !head.windows(4).any(|w| w == b"\r\n\r\n") { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => head.extend_from_slice(&buf[..n]), + } + } + let body = slow_body(); + let head = format!( + "HTTP/1.1 200 OK\r\ncache-control: max-age=60\r\ncontent-length: {}\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + match first_body { + None => { + let _ = stream.write_all(&body).await; + } + Some(FirstBody::Trickle) => { + let _ = stream.write_all(&body[..5]).await; + for chunk in body[5..].chunks(CHUNK) { + actix_web::rt::time::sleep(Duration::from_millis(100)).await; + if stream.write_all(chunk).await.is_err() { + return; + } + } + } + Some(FirstBody::Stall) => { + let _ = stream.write_all(&body[..5]).await; + // hold the connection open until the proxy closes it + while matches!(stream.read(&mut buf).await, Ok(n) if n > 0) {} + } + } +} + +/// sends a GET for `/page` on a raw connection and returns it once the +/// start of the body has arrived. +async fn raw_leader(server: &Running) -> TcpStream { + let mut leader = TcpStream::connect(server.addr).await.unwrap(); + // the same host as the followers', which is part of the cache key + let request = format!("GET /page HTTP/1.1\r\nhost: {}\r\n\r\n", server.addr); + leader.write_all(request.as_bytes()).await.unwrap(); + let mut received = Vec::new(); + let mut buf = [0; 1024]; + while !received.ends_with(b"first") { + let n = timeout(PATIENCE, leader.read(&mut buf)) + .await + .unwrap() + .unwrap(); + assert!(n > 0, "the proxy closed the leader's connection"); + received.extend_from_slice(&buf[..n]); + } + leader +} + +/// starts `FOLLOWERS` GETs for `/page` and gives them time to join the +/// leader's flight. no event shows that a follower is waiting, and a +/// follower that joins late goes to the origin and still passes. +async fn spawn_followers( + client: &HttpClient, + server: &Running, +) -> actix_web::rt::task::JoinHandle> { + let (client, url) = (client.clone(), server.url("/page")); + let followers = actix_web::rt::spawn(async move { + join_all((0..FOLLOWERS).map(|_| fetch(&client, &url, &[]))).await + }); + actix_web::rt::time::sleep(Duration::from_millis(300)).await; + followers +} + +fn assert_own_whole_responses(followers: &[Answer]) { + for answer in followers { + assert_eq!((answer.status, answer.cache.as_str()), (200, "MISS")); + assert!(answer.body == slow_body(), "follower got a different body"); + } +} + +#[actix_web::test] +async fn followers_are_released_when_the_leaders_client_disconnects() { + let (origin_url, connections) = origin_slowing_the_first_body(FirstBody::Trickle).await; + let server = common::spawn_workers(&origin_url, WORKERS, |_| {}); + let client = common::client(); + + let leader = raw_leader(&server).await; + let followers = spawn_followers(&client, &server).await; + let disconnected = Instant::now(); + drop(leader); + let followers = followers.await.unwrap(); + + // the leader's body would take over 6 seconds to arrive, and the + // upstream timeout is 30 seconds + assert!( + disconnected.elapsed() < Duration::from_secs(3), + "followers waited {:?} after the disconnect", + disconnected.elapsed() + ); + assert_own_whole_responses(&followers); + assert_eq!(connections.load(Ordering::SeqCst), 1 + FOLLOWERS); + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; +} + +#[actix_web::test] +async fn followers_of_a_stalled_leader_stop_waiting_after_the_upstream_timeout() { + let (origin_url, connections) = origin_slowing_the_first_body(FirstBody::Stall).await; + let server = common::spawn_workers(&origin_url, WORKERS, |c| { + c.upstream_timeout_seconds = 2; + }); + let client = common::client(); + + let _leader = raw_leader(&server).await; + let started = Instant::now(); + let followers = spawn_followers(&client, &server).await.await.unwrap(); + + assert!( + started.elapsed() >= Duration::from_millis(1500), + "followers did not wait for the leader" + ); + assert!( + started.elapsed() < Duration::from_secs(5), + "followers waited {:?}", + started.elapsed() + ); + assert_own_whole_responses(&followers); + assert_eq!(connections.load(Ordering::SeqCst), 1 + FOLLOWERS); + server.stop().await; +} + +#[actix_web::test] +async fn requests_that_bypass_the_cache_do_not_coalesce() { + let origin = origin_responding(cacheable("body")).await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + let cases = [ + ("/authorization", ("authorization", "Bearer token")), + ("/cookie", ("cookie", "session=1")), + ("/no-cache", ("cache-control", "no-cache")), + ("/if-none-match", ("if-none-match", "\"v1\"")), + ]; + + for (n, (path, field)) in cases.into_iter().enumerate() { + let before = 2 * n; + let leader = lead(&client, &server, path, &[]); + origin_receives(&origin, before + 1).await; + let follower = fetch(&client, &server.url(path), &[field]).await; + let leader = leader.await.unwrap(); + + assert_eq!(leader.cache, "MISS"); + assert_eq!(follower.cache, "MISS", "{field:?} coalesced"); + assert_eq!(origin_requests(&origin).await, before + 2, "{field:?}"); + } + assert_eq!(health(&client, &server).await["cache"]["coalesced"], 0); + server.stop().await; +} + +#[actix_web::test] +async fn concurrent_stale_requests_send_one_conditional_request() { + let origin = MockServer::start().await; + Mock::given(header("if-none-match", "\"v1\"")) + .respond_with( + ResponseTemplate::new(304) + .insert_header("cache-control", "max-age=60") + .insert_header("etag", "\"v1\"") + .set_delay(ORIGIN_DELAY), + ) + .mount(&origin) + .await; + Mock::given(any()) + .respond_with( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=1") + .insert_header("etag", "\"v1\"") + .set_body_string("stored"), + ) + .mount(&origin) + .await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + fetch(&client, &server.url("/page"), &[]).await; + actix_web::rt::time::sleep(Duration::from_millis(1100)).await; + + let leader = lead(&client, &server, "/page", &[]); + origin_receives(&origin, 2).await; + let followers = fetch_many(&client, &server, "/page", &[], FOLLOWERS).await; + let leader = leader.await.unwrap(); + + assert_eq!( + (leader.cache.as_str(), leader.body.as_ref()), + ("REVALIDATED", &b"stored"[..]) + ); + for follower in &followers { + assert_eq!(follower.status, 200); + assert_eq!(follower.body, "stored"); + } + assert_eq!(cache_statuses(&followers), ["COALESCED"; FOLLOWERS]); + assert_eq!(origin_requests(&origin).await, 2); + server.stop().await; +} + +#[actix_web::test] +async fn different_keys_do_not_wait_for_each_other() { + let origin = MockServer::start().await; + Mock::given(path("/slow")) + .respond_with(cacheable("slow").set_delay(Duration::from_secs(4))) + .mount(&origin) + .await; + Mock::given(path("/fast")) + .respond_with(cacheable("fast")) + .mount(&origin) + .await; + let server = common::spawn_workers(&origin.uri(), WORKERS, |_| {}); + let client = common::client(); + + let slow = lead(&client, &server, "/slow", &[]); + origin_receives(&origin, 1).await; + let started = Instant::now(); + let fast_leader = lead(&client, &server, "/fast", &[]); + origin_receives(&origin, 2).await; + let fast = fetch_many(&client, &server, "/fast", &[], FOLLOWERS).await; + let fast_leader = fast_leader.await.unwrap(); + + assert!( + started.elapsed() < Duration::from_secs(3), + "/fast waited {:?} for /slow", + started.elapsed() + ); + assert_eq!(fast_leader.cache, "MISS"); + assert_eq!(cache_statuses(&fast), ["COALESCED"; FOLLOWERS]); + assert_eq!(slow.await.unwrap().body, "slow"); + assert_eq!(origin_requests(&origin).await, 2); + server.stop().await; +} diff --git a/tests/integration/main.rs b/tests/integration/main.rs index 9552396..e07e50a 100644 --- a/tests/integration/main.rs +++ b/tests/integration/main.rs @@ -1,5 +1,6 @@ mod body_failure; mod cache; +mod coalescing; mod common; mod forwarded; mod http2; From d5183297e60a4666e8b84b8b8c6a0c0179233b4b Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:36:18 +0100 Subject: [PATCH 3/3] docs(readme): describe request coalescing --- README.md | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 6400c76..265fe03 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,7 @@ Responses served from the cache carry `Age`. Proxied and asset responses carry ` - `MISS`: the origin's response. - `REVALIDATED`: the origin answered a conditional request with `304 Not Modified`, and the client got the stored body with the 304's header fields. - `STALE`: a stale stored response, served under `stale-while-revalidate` or `stale-if-error`. +- `COALESCED`: a response that a concurrent request for the same key stored while this request waited for it. See [Request coalescing](#request-coalescing). `--cache-ttl-seconds` (default 300) caps the freshness lifetime of any entry, whatever the origin sent. `--cache-size-mb` (default 100) bounds origin responses and assets together, measured in bytes. The largest single entry is 8 MiB or the cache size, whichever is smaller. Larger bodies stream to the client without being stored. Setting either option to 0 turns caching off. @@ -91,6 +92,16 @@ A `GET` that finds a stale response sends the origin `If-None-Match` from the st A response with `must-revalidate`, `proxy-revalidate` or `s-maxage` is never served stale, whatever its stale windows. If the origin cannot be reached to revalidate it, the client gets `504 Gateway Timeout`. A request with `Cache-Control: max-age` never gets a stale response. `HEAD` requests are answered only from fresh responses. +### Request coalescing + +When several requests miss on the same key at once, the first one, the leader, goes to the origin and the others wait for it. The key is the scheme, host, path and query, plus the request's values for the `Vary` field names of the latest stored response for that URL, if the cache has one. Requests for different known variants therefore do not wait for each other. Requests that find a stale response they must revalidate coalesce the same way, so the origin gets one conditional request. + +- Only `GET` and `HEAD` requests that may be served from the cache wait. A request with `Authorization`, `Cookie`, a conditional header, request `Cache-Control: no-cache` or `no-store`, or a method-override header goes to the origin as before. +- Only a `GET` leads. A `HEAD` request with no `GET` in flight goes to the origin. +- The leader's response streams to its client as usual. The waiting requests look up the cache once the response is stored, or once the proxy knows it will not be stored: it is not storable, its body passes the entry size limit or fails, the leader's client disconnects, or the origin fails or times out. +- A waiting request gets the stored response with `X-Shadowstep-Cache: COALESCED`. If the cache has nothing it may use, for example because the response was `private` or had `Vary` values that differ from the waiting request's, the waiting request goes to the origin on its own. One client's uncacheable response never goes to another client. +- A request waits for at most `--upstream-timeout-seconds`, then goes to the origin on its own. + Assets share the same cache. A stored asset is read from disk again when the file's size or modified time changes. `/health` counts responses for proxied requests and assets together. Each response counts once: @@ -99,14 +110,16 @@ Assets share the same cache. A stored asset is read from disk again when the fil - `misses`: responses from the origin, including errors. - `revalidations`: 304s that freshened a stored response, in the foreground or the background. - `stale`: stale responses served under `stale-while-revalidate` or `stale-if-error`. +- `coalesced`: responses stored by a concurrent request that this request waited for. -`background_refreshes` counts background revalidations started. `hit_ratio` is the share of responses whose body came from the cache: hits, revalidations and stale responses. `items` and `bytes` describe the whole cache. +`background_refreshes` counts background revalidations started. `hit_ratio` is the share of responses whose body came from the cache: hits, revalidations, stale and coalesced responses. `items` and `bytes` describe the whole cache. Known limits: - A response with `Cache-Control: no-cache` is not stored, although RFC 9111 allows storing it and revalidating it on every use. - A stored response is never used to answer a client's conditional request with a 304. A fresh hit always gets the full response. -- There is no request coalescing. Concurrent misses for the same URL all go to the origin. +- Requests with `Cookie` do not coalesce. If browsers send a cookie with every request to the site, only cookieless clients coalesce. +- The proxy notices that a client has disconnected only when it next writes to it. If the leader's client disconnects while the origin has stopped sending the body, the waiting requests go to the origin after `--upstream-timeout-seconds`. - The host is part of the key, so a client that sends many different `Host` values can create many entries. The byte bound on the cache still applies. - Each process has its own cache. Replicas do not share entries or invalidations.