From 59372f1ca1ca686d540ce8ca5f25600d915d2b83 Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:01:18 +0100 Subject: [PATCH 1/3] feat(cache): revalidate and serve stale responses A stale response with an ETag or Last-Modified stays stored for up to --cache-ttl-seconds more, so the proxy can revalidate it with a conditional request. Responses with s-maxage now count as proxy-revalidate, as RFC 9111 section 5.2.2.10 requires, so they are never served stale. --- src/cache.rs | 324 +++++++++++++++++-- src/lib.rs | 38 ++- src/proxy.rs | 389 ++++++++++++++++++---- tests/integration/main.rs | 1 + tests/integration/revalidation.rs | 514 ++++++++++++++++++++++++++++++ 5 files changed, 1173 insertions(+), 93 deletions(-) create mode 100644 tests/integration/revalidation.rs diff --git a/src/cache.rs b/src/cache.rs index 9ea39e2..eed6f14 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -6,9 +6,10 @@ use actix_web::http::{Method, StatusCode}; use bytes::Bytes; use moka::sync::Cache; use moka::Expiry; +use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant, SystemTime}; const MIB: u64 = 1024 * 1024; @@ -91,7 +92,11 @@ pub struct StoredResponse { pub body: Bytes, stored_at: Instant, initial_age: Duration, + /// the freshness lifetime left when the response was stored ttl: Duration, + /// how long the store keeps the response once it is stale + grace: Duration, + stale: StaleRules, } impl StoredResponse { @@ -103,6 +108,132 @@ impl StoredResponse { fn is_fresh(&self) -> bool { self.stored_at.elapsed() < self.ttl } + + /// how long the response has been stale, or `None` while it is fresh. + fn staleness(&self) -> Option { + self.stored_at.elapsed().checked_sub(self.ttl) + } + + fn stale_within(&self, window: Duration) -> bool { + !self.stale.must_revalidate && self.staleness().is_some_and(|s| s < window) + } + + /// whether the response may be served stale while a background request + /// revalidates it (RFC 5861 section 3). + pub fn may_serve_while_revalidating(&self) -> bool { + self.stale_within(self.stale.while_revalidate) + } + + /// whether the response may be served stale in place of an origin error + /// (RFC 5861 section 4). + pub fn may_serve_on_error(&self) -> bool { + self.stale_within(self.stale.if_error) + } + + /// whether a stale use of the response needs a successful revalidation + /// (RFC 9111 sections 5.2.2.2, 5.2.2.8 and 5.2.2.10). + pub fn must_revalidate(&self) -> bool { + self.stale.must_revalidate + } + + /// the conditional request fields built from the stored validators (RFC + /// 9111 section 4.3.1). + pub fn validators(&self) -> Fields { + let mut fields = Fields::new(); + for (name, value) in &self.headers { + if name == header::ETAG { + fields.push((header::IF_NONE_MATCH, value.clone())); + } else if name == header::LAST_MODIFIED { + fields.push((header::IF_MODIFIED_SINCE, value.clone())); + } + } + fields + } +} + +/// what a response's Cache-Control allows once it is stale. +#[derive(Clone, Copy, Default)] +struct StaleRules { + while_revalidate: Duration, + if_error: Duration, + must_revalidate: bool, +} + +impl StaleRules { + fn parse(directives: &Directives) -> Self { + StaleRules { + while_revalidate: directives + .seconds("stale-while-revalidate") + .unwrap_or_default(), + if_error: directives.seconds("stale-if-error").unwrap_or_default(), + // s-maxage carries the semantics of proxy-revalidate for a shared + // cache (RFC 9111 section 5.2.2.10) + must_revalidate: ["must-revalidate", "proxy-revalidate", "s-maxage"] + .iter() + .any(|d| directives.has(d)), + } + } + + /// how long a stale response stays in the store: long enough for its + /// stale windows, and up to `cap` when it has a validator, so that it can + /// be revalidated instead of fetched again in full. never more than `cap`. + fn grace(&self, has_validator: bool, cap: Duration) -> Duration { + let windows = if self.must_revalidate { + Duration::ZERO + } else { + self.while_revalidate.max(self.if_error) + }; + let revalidation = if has_validator { cap } else { Duration::ZERO }; + windows.max(revalidation).min(cap) + } +} + +/// a stored response for a request, and whether it may answer it as is. +pub enum Lookup { + /// fresh, and no older than the request allows + Fresh(Arc), + /// stale, or older than the request's max-age, so it needs revalidation + /// before it is used, unless a stale window applies + Stale(StaleEntry), +} + +/// a stored response that needs revalidation, and where it is stored. +pub struct StaleEntry { + pub response: Arc, + key: ResponseKey, +} + +/// the fields of a stored response updated from a 304 (RFC 9111 section +/// 3.2): each field in `update` replaces the stored values of that name. +/// `update` must already leave out the fields that are not stored. +pub fn updated_fields(stored: &Fields, update: &Fields) -> Fields { + let mut fields: Fields = stored + .iter() + .filter(|(name, _)| !update.iter().any(|(n, _)| n == name)) + .cloned() + .collect(); + fields.extend(update.iter().cloned()); + fields +} + +/// the only background revalidation of a stored response. dropping it lets +/// the next stale request start another. +pub struct RefreshGuard { + refreshing: Arc>>, + key: ResponseKey, +} + +impl Drop for RefreshGuard { + fn drop(&mut self) { + lock(&self.refreshing).remove(&self.key); + } +} + +/// the set of keys being refreshed. a panic while the lock is held cannot +/// leave the set inconsistent, so a poisoned lock is still usable. +fn lock(set: &Mutex>) -> std::sync::MutexGuard<'_, HashSet> { + set.lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) } fn entry_weight(key: &Key, entry: &Entry) -> u32 { @@ -152,7 +283,7 @@ impl Expiry for EntryExpiry { fn entry_ttl(entry: &Entry) -> Duration { match entry { Entry::Asset(asset) => asset.ttl, - Entry::Response(response) => response.ttl, + Entry::Response(response) => response.ttl.saturating_add(response.grace), } } @@ -162,6 +293,7 @@ pub struct Store { entries: Cache, index: Cache>, generation: Arc, + refreshing: Arc>>, ttl: Duration, max_entry: u64, } @@ -176,11 +308,13 @@ impl Store { .weigher(entry_weight) .expire_after(EntryExpiry) .build(), + // an entry lives for at most its ttl plus a grace of up to ttl index: Cache::builder() .max_capacity(MAX_INDEXED_URIS) - .time_to_live(ttl.max(Duration::from_secs(1))) + .time_to_live(ttl.saturating_mul(2).max(Duration::from_secs(1))) .build(), generation: Arc::new(AtomicU64::new(0)), + refreshing: Arc::default(), ttl, max_entry: capacity.min(MAX_ENTRY_BYTES), } @@ -234,34 +368,118 @@ impl Store { asset } - /// a fresh stored response for `primary` that matches the request's - /// `Vary` fields (RFC 9111 section 4.1) and is no older than `max_age`. + /// the stored response for `primary` that matches the request's `Vary` + /// fields (RFC 9111 section 4.1). it is fresh when its freshness lifetime + /// has not run out and it is no older than `max_age`. pub fn lookup( &self, primary: &PrimaryKey, request: &HeaderMap, max_age: Option, - ) -> Option> { + ) -> Option { let index = self.index.get(primary)?; - let key = Key::Response(ResponseKey { + let key = ResponseKey { primary: primary.clone(), generation: index.generation, vary: vary_values(&index.vary, request), - }); - let Entry::Response(response) = self.entries.get(&key)? else { - return None; }; - if !response.is_fresh() { + let Entry::Response(response) = self.entries.get(&Key::Response(key.clone()))? else { return None; + }; + if response.is_fresh() && max_age.is_none_or(|max_age| response.age() <= max_age) { + return Some(Lookup::Fresh(response)); } - if max_age.is_some_and(|max_age| response.age() > max_age) { - return None; + Some(Lookup::Stale(StaleEntry { response, key })) + } + + /// claims the background revalidation of `entry`, or `None` when one is + /// already running. + pub fn start_refresh(&self, entry: &StaleEntry) -> Option { + lock(&self.refreshing) + .insert(entry.key.clone()) + .then(|| RefreshGuard { + refreshing: self.refreshing.clone(), + key: entry.key.clone(), + }) + } + + /// `entry` updated from a 304 (RFC 9111 section 4.3.4). `update` holds + /// the 304's fields that are stored, and `age` its `Age`. the updated + /// response replaces the entry when it may still be stored, and the + /// entry is removed when it may not. an entry that another response has + /// replaced since the lookup is left alone. + pub fn freshen( + &self, + entry: &StaleEntry, + request_policy: &RequestPolicy, + request: &HeaderMap, + update: &Fields, + age: Option<&HeaderValue>, + ) -> Arc { + let stored = &entry.response; + let headers = updated_fields(&stored.headers, update); + let mut map = HeaderMap::with_capacity(headers.len() + 1); + for (name, value) in &headers { + map.append(name.clone(), value.clone()); } - Some(response) + if let Some(age) = age { + map.insert(header::AGE, age.clone()); + } + let storable = storable(request_policy, request, stored.status, &map); + let key = Key::Response(entry.key.clone()); + let current = + matches!(self.entries.get(&key), Some(Entry::Response(r)) if Arc::ptr_eq(&r, stored)); + let policy = match storable { + Some((policy, vary)) if current => { + let body = stored.body.clone(); + let primary = entry.key.primary.clone(); + if let Some(fresh) = self.insert_response( + primary, + vary, + &policy, + stored.status, + headers.clone(), + body, + ) { + return fresh; + } + policy + } + Some((policy, _)) => policy, + None => { + if current { + self.entries.invalidate(&key); + } + Storable::default() + } + }; + Arc::new(self.response(&policy, stored.status, headers, stored.body.clone())) + } + + fn response( + &self, + policy: &Storable, + status: StatusCode, + headers: Fields, + body: Bytes, + ) -> StoredResponse { + let mut response = StoredResponse { + status, + headers, + body, + stored_at: Instant::now(), + initial_age: policy.initial_age, + ttl: policy.ttl.min(self.ttl), + grace: Duration::ZERO, + stale: policy.stale, + }; + let has_validator = !response.validators().is_empty(); + response.grace = policy.stale.grace(has_validator, self.ttl); + response } /// stores `body` under `primary` and the request's values for the - /// response's `Vary` field names. + /// response's `Vary` field names, and returns the stored response. pub fn insert_response( &self, primary: PrimaryKey, @@ -270,9 +488,9 @@ impl Store { status: StatusCode, headers: Fields, body: Bytes, - ) { + ) -> Option> { if !self.enabled() || body.len() as u64 > self.max_entry { - return; + return None; } // a response whose Vary names differ from the stored one's replaces // every stored variant @@ -292,17 +510,10 @@ impl Store { generation, vary: vary.values, }); - let response = StoredResponse { - status, - headers, - body, - stored_at: Instant::now(), - initial_age: policy.initial_age, - ttl: policy.ttl.min(self.ttl), - }; - self.entries - .insert(key, Entry::Response(Arc::new(response))); + let response = Arc::new(self.response(policy, status, headers, body)); + self.entries.insert(key, Entry::Response(response.clone())); self.entries.run_pending_tasks(); + Some(response) } /// makes every stored response for `primary` unreachable (RFC 9111 @@ -426,8 +637,20 @@ pub struct RequestPolicy { /// whether the origin's response may be stored may_store: bool, has_credentials: bool, + /// whether the request has its own preconditions (RFC 9110 section 13.1) + pub conditional: bool, } +/// the request fields that make a request conditional (RFC 9110 section +/// 13.1). +pub const PRECONDITION_HEADERS: [HeaderName; 5] = [ + header::IF_MATCH, + header::IF_NONE_MATCH, + header::IF_MODIFIED_SINCE, + header::IF_UNMODIFIED_SINCE, + header::IF_RANGE, +]; + impl RequestPolicy { pub fn new(method: &Method, headers: &HeaderMap) -> Self { let directives = Directives::parse(headers.get_all(header::CACHE_CONTROL)); @@ -445,14 +668,24 @@ impl RequestPolicy { max_age: directives.seconds("max-age"), may_store: cacheable_method && *method == Method::GET && !no_store, has_credentials: headers.contains_key(AUTHORIZATION) || headers.contains_key(COOKIE), + conditional: PRECONDITION_HEADERS.iter().any(|h| headers.contains_key(h)), } } + + /// whether a stale response may answer the request. a request max-age + /// asks for a response no older than that, so it rules out stale ones + /// (RFC 9111 section 5.2.1.1). + pub fn may_serve_stale(&self) -> bool { + self.max_age.is_none() + } } /// a response that the store may keep, and for how long. +#[derive(Default)] pub struct Storable { ttl: Duration, initial_age: Duration, + stale: StaleRules, } /// status codes that are cacheable by default (RFC 9110 section 15.1). @@ -462,7 +695,7 @@ const CACHEABLE_BY_DEFAULT: [u16; 11] = [200, 203, 204, 300, 301, 308, 404, 405, /// Cache-Control directives that stop a shared cache storing a response. a /// shared cache must not store private responses, qualified or not (RFC 9111 /// section 5.2.2.7). no-cache would need revalidation on every use (section -/// 5.2.2.4), and this cache does not revalidate. +/// 5.2.2.4), and this cache revalidates only stale responses. const UNSTORABLE_DIRECTIVES: [&str; 3] = ["no-store", "private", "no-cache"]; /// whether a shared cache may store the origin's response to a request @@ -499,7 +732,12 @@ pub fn storable( } let values = vary_values(&names, request); - Some((Storable { ttl, initial_age }, Vary { names, values })) + let storable = Storable { + ttl, + initial_age, + stale: StaleRules::parse(&directives), + }; + Some((storable, Vary { names, values })) } /// the response's Cache-Control directives, or `None` when the response @@ -620,6 +858,34 @@ mod tests { assert_eq!(d.seconds("max-age"), Some(Duration::from_secs(7))); } + #[test] + fn not_modified_fields_replace_stored_fields_of_the_same_name() { + let field = |name: &'static str, value: &'static str| { + ( + HeaderName::from_static(name), + HeaderValue::from_static(value), + ) + }; + let stored = vec![ + field("x-version", "1"), + field("cache-control", "max-age=1"), + field("x-version", "1b"), + field("etag", "\"v1\""), + ]; + let update = vec![field("cache-control", "max-age=60"), field("x-new", "yes")]; + + assert_eq!( + updated_fields(&stored, &update), + vec![ + field("x-version", "1"), + field("x-version", "1b"), + field("etag", "\"v1\""), + field("cache-control", "max-age=60"), + field("x-new", "yes"), + ] + ); + } + #[test] fn non_ascii_field_value_is_malformed() { let value = HeaderValue::from_bytes(b"max-age=60, \xff").unwrap(); diff --git a/src/lib.rs b/src/lib.rs index b7de6c3..5c6814e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -35,11 +35,16 @@ use url::Url; use crate::cache::Store; use crate::config::Config; -/// hits and misses across origin responses and local assets +/// 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. #[derive(Default)] pub struct CacheStats { hits: AtomicU64, misses: AtomicU64, + revalidations: AtomicU64, + stale: AtomicU64, + background_refreshes: AtomicU64, } impl CacheStats { @@ -50,6 +55,20 @@ impl CacheStats { fn miss(&self) { self.misses.fetch_add(1, Ordering::Relaxed); } + + /// a 304 that freshened a stored response, in the foreground or the + /// background + fn revalidation(&self) { + self.revalidations.fetch_add(1, Ordering::Relaxed); + } + + fn stale(&self) { + self.stale.fetch_add(1, Ordering::Relaxed); + } + + fn background_refresh(&self) { + self.background_refreshes.fetch_add(1, Ordering::Relaxed); + } } /// application state, including cache @@ -64,17 +83,26 @@ pub struct AppState { #[get("/health")] async fn health_check(state: web::Data) -> impl Responder { - let hits = state.cache_stats.hits.load(Ordering::Relaxed); - let misses = state.cache_stats.misses.load(Ordering::Relaxed); + let stats = &state.cache_stats; + let hits = stats.hits.load(Ordering::Relaxed); + let misses = stats.misses.load(Ordering::Relaxed); + let revalidations = stats.revalidations.load(Ordering::Relaxed); + let stale = stats.stale.load(Ordering::Relaxed); + // the share of responses whose body came from the cache + let from_cache = hits + revalidations + stale; + let total = from_cache + misses; HttpResponse::Ok().json(serde_json::json!({ "status": "ok", "cache": { "hits": hits, "misses": misses, + "revalidations": revalidations, + "stale": stale, + "background_refreshes": stats.background_refreshes.load(Ordering::Relaxed), "items": state.cache.entry_count(), "bytes": state.cache.weighted_size(), - "hit_ratio": if hits + misses > 0 { - hits as f64 / (hits + misses) as f64 + "hit_ratio": if total > 0 { + from_cache as f64 / total as f64 } else { 0.0 } diff --git a/src/proxy.rs b/src/proxy.rs index 1886900..8f52de3 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -1,12 +1,12 @@ use actix_web::body::SizedStream; use actix_web::error::PayloadError; use actix_web::http::header::{self, HeaderMap, HeaderName, HeaderValue}; -use actix_web::http::StatusCode; +use actix_web::http::{Method, StatusCode}; use actix_web::{web, HttpRequest, HttpResponse}; use bytes::{Bytes, BytesMut}; use futures_util::{Stream, StreamExt}; use http_body_util::combinators::UnsyncBoxBody; -use http_body_util::{BodyExt, Empty, StreamBody}; +use http_body_util::{BodyExt, Empty, Limited, StreamBody}; use hyper::body::{Frame, Incoming}; use hyper::{Request as HyperRequest, Uri}; use log::{debug, error, warn}; @@ -16,7 +16,10 @@ use std::task::{Context, Poll}; use tokio::sync::mpsc; use url::{Position, Url}; -use crate::cache::{self, Fields, PrimaryKey, RequestPolicy, StoredResponse}; +use crate::cache::{ + self, Fields, Lookup, PrimaryKey, RequestPolicy, StaleEntry, StoredResponse, + PRECONDITION_HEADERS, +}; use crate::forwarded::{ClientInfo, CLIENT_FORWARDING_HEADERS, URL_OVERRIDE_HEADERS}; use crate::AppState; @@ -46,35 +49,46 @@ 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()); - if let Some(hit) = serve_from_cache(&req, &state, &cache_key, &request_policy) { - return hit; - } - state.cache_stats.miss(); + 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 Some((hyper_req_builder, target_uri)) = - upstream_request(&req, &state, &client, path_and_query) + let Some((builder, target_uri)) = upstream_request(&req, &state, &client, path_and_query) else { - return HttpResponse::InternalServerError().body("internal server error"); + state.cache_stats.miss(); + return error_response(StatusCode::INTERNAL_SERVER_ERROR); }; - - let body = match request_body(payload).await { - Ok(body) => body, - Err(e) => { - warn!("Failed to read request body: {}", e); - return HttpResponse::BadRequest().body("bad request"); - } + // a client's own preconditions go to the origin unchanged, and then the + // origin's answer is for the client, not for the stored response + let revalidating = stale.as_ref().filter(|_| !request_policy.conditional); + let builder = match revalidating { + Some(entry) => with_validators(builder, &entry.response), + None => builder, + }; + let exchange = Exchange { + revalidating: revalidating.is_some(), + req, + state, + cache_key, + request_policy, + stale, }; - let hyper_req = match hyper_req_builder.body(body) { - Ok(req) => req, - Err(e) => { - error!("Failed to build hyper request: {}", e); - return HttpResponse::InternalServerError().body("internal server error"); + let hyper_req = match origin_request(builder, payload).await { + Ok(hyper_req) => hyper_req, + Err(status) => { + exchange.state.cache_stats.miss(); + return error_response(status); } }; - let upstream = - tokio::time::timeout(state.upstream_timeout, state.http_client.request(hyper_req)).await; + let upstream = tokio::time::timeout( + exchange.state.upstream_timeout, + exchange.state.http_client.request(hyper_req), + ) + .await; match upstream { Ok(Ok(upstream_response)) => { @@ -82,42 +96,285 @@ pub async fn forward_to_upstream( "Received response from upstream: {:?}", upstream_response.status() ); - let (parts, body) = upstream_response.into_parts(); - let head = Head::from_upstream(&parts); - let store = store_plan(&req, &state, cache_key, &request_policy, &head); - client_response(head, body, store) + exchange.answered(upstream_response) } Ok(Err(e)) => { error!("Error forwarding request to upstream {}: {}", target_uri, e); - HttpResponse::BadGateway().body("bad gateway") + exchange.failed(StatusCode::BAD_GATEWAY) } Err(_) => { error!( "Upstream {} sent no response headers within {:?}", - target_uri, state.upstream_timeout + target_uri, exchange.state.upstream_timeout ); - HttpResponse::GatewayTimeout().body("gateway timeout") + exchange.failed(StatusCode::GATEWAY_TIMEOUT) } } } -/// the client's response from the cache, when a fresh stored response may -/// answer this request. -fn serve_from_cache( +/// what the cache can do for a request before it goes to the origin. +enum Cached { + /// the cache answers the request + Answer(HttpResponse), + /// the request goes to the origin to revalidate this stored response + Stale(StaleEntry), + /// the request goes to the origin + Nothing, +} + +/// 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. +fn from_cache( req: &HttpRequest, - state: &AppState, + state: &web::Data, cache_key: &PrimaryKey, request_policy: &RequestPolicy, -) -> Option { +) -> Cached { if !request_policy.may_serve { - return None; + return Cached::Nothing; } - let stored = state + match state .cache - .lookup(cache_key, req.headers(), request_policy.max_age)?; - debug!("Cache hit for {} {}", req.method(), req.uri()); - state.cache_stats.hit(); - Some(stored_response(&stored)) + .lookup(cache_key, req.headers(), request_policy.max_age) + { + Some(Lookup::Fresh(stored)) => { + debug!("Cache hit for {} {}", req.method(), req.uri()); + state.cache_stats.hit(); + Cached::Answer(cached_response(&stored, "HIT")) + } + Some(Lookup::Stale(entry)) if req.method() == Method::GET => { + while_revalidating(req, state, cache_key, request_policy, entry) + } + _ => Cached::Nothing, + } +} + +/// the stale response and a background revalidation when the response's +/// stale-while-revalidate window allows it (RFC 5861 section 3), or else +/// the entry to revalidate before use. +fn while_revalidating( + req: &HttpRequest, + state: &web::Data, + cache_key: &PrimaryKey, + request_policy: &RequestPolicy, + entry: StaleEntry, +) -> Cached { + if !request_policy.may_serve_stale() || !entry.response.may_serve_while_revalidating() { + return Cached::Stale(entry); + } + debug!("Serving stale {} while revalidating", req.uri()); + state.cache_stats.stale(); + let response = cached_response(&entry.response, "STALE"); + refresh_in_background(req, state, cache_key.clone(), entry); + Cached::Answer(response) +} + +/// revalidates `entry` in a task on this worker, unless another task is +/// already revalidating it. the origin request is built as for a miss, with +/// the stored validators in place of the client's preconditions. +fn refresh_in_background( + req: &HttpRequest, + state: &web::Data, + cache_key: PrimaryKey, + entry: StaleEntry, +) { + let Some(guard) = state.cache.start_refresh(&entry) else { + return; + }; + let client = ClientInfo::from_request(req); + let path_and_query = req.uri().path_and_query().map_or("", |pq| pq.as_str()); + let Some((mut builder, target_uri)) = upstream_request(req, state, &client, path_and_query) + else { + return; + }; + if let Some(headers) = builder.headers_mut() { + for name in &PRECONDITION_HEADERS { + headers.remove(name.as_str()); + } + } + let Ok(hyper_req) = with_validators(builder, &entry.response).body(empty_body()) else { + return; + }; + state.cache_stats.background_refresh(); + let (req, state) = (req.clone(), state.clone()); + actix_web::rt::spawn(async move { + let _guard = guard; + let upstream = + tokio::time::timeout(state.upstream_timeout, state.http_client.request(hyper_req)) + .await; + match upstream { + Ok(Ok(response)) => refresh(&req, &state, cache_key, &entry, response).await, + Ok(Err(e)) => warn!("Background revalidation of {} failed: {}", target_uri, e), + Err(_) => warn!("Background revalidation of {} timed out", target_uri), + } + }); +} + +/// updates `entry` from the origin's answer to a background revalidation: a +/// 304 freshens it, and a storable response replaces it. anything else +/// leaves it for its stale windows. +async fn refresh( + req: &HttpRequest, + state: &AppState, + cache_key: PrimaryKey, + entry: &StaleEntry, + response: hyper::Response, +) { + let (parts, body) = response.into_parts(); + let head = Head::from_upstream(&parts); + let request_policy = RequestPolicy::new(req.method(), req.headers()); + if head.status == StatusCode::NOT_MODIFIED { + state.cache.freshen( + entry, + &request_policy, + req.headers(), + &stored_fields(forwardable_fields(&head)), + head.map.get(header::AGE), + ); + state.cache_stats.revalidation(); + return; + } + let Some(store) = store_plan(req, state, cache_key, &request_policy, &head) else { + return; + }; + let limit = usize::try_from(store.limit).unwrap_or(usize::MAX); + match Limited::new(body, limit).collect().await { + Ok(collected) => (store.finish)( + head.status, + stored_fields(forwardable_fields(&head)), + collected.to_bytes(), + ), + Err(e) => debug!("Background revalidation body not stored: {}", e), + } +} + +/// statuses for which stale-if-error serves a stale response in place of +/// the origin's (RFC 5861 section 4). +const ERROR_STATUSES: [StatusCode; 4] = [ + StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::BAD_GATEWAY, + StatusCode::SERVICE_UNAVAILABLE, + StatusCode::GATEWAY_TIMEOUT, +]; + +/// one request on its way to the origin, and the stored response it may +/// revalidate. +struct Exchange { + req: HttpRequest, + state: web::Data, + cache_key: PrimaryKey, + request_policy: RequestPolicy, + stale: Option, + /// whether the origin request carries the stale response's validators + revalidating: bool, +} + +impl Exchange { + /// the client's response once the origin has answered. + fn answered(self, response: hyper::Response) -> HttpResponse { + let (parts, body) = response.into_parts(); + let head = Head::from_upstream(&parts); + if let Some(stale) = &self.stale { + if self.revalidating && head.status == StatusCode::NOT_MODIFIED { + return self.revalidated(stale, &head); + } + if ERROR_STATUSES.contains(&head.status) && self.may_serve_on_error(stale) { + return self.serve_stale(stale); + } + } + self.state.cache_stats.miss(); + let store = store_plan( + &self.req, + &self.state, + self.cache_key, + &self.request_policy, + &head, + ); + client_response(head, body, store) + } + + /// the client's response when the origin could not be reached or sent no + /// response headers in time. a response that must be revalidated gets a + /// 504 (RFC 9111 section 5.2.2.2). + fn failed(self, status: StatusCode) -> HttpResponse { + let mut status = status; + if let Some(stale) = &self.stale { + if self.may_serve_on_error(stale) { + return self.serve_stale(stale); + } + if stale.response.must_revalidate() { + status = StatusCode::GATEWAY_TIMEOUT; + } + } + self.state.cache_stats.miss(); + error_response(status) + } + + fn may_serve_on_error(&self, stale: &StaleEntry) -> bool { + self.request_policy.may_serve_stale() && stale.response.may_serve_on_error() + } + + fn serve_stale(&self, stale: &StaleEntry) -> HttpResponse { + debug!( + "Serving stale {} in place of an origin error", + self.req.uri() + ); + self.state.cache_stats.stale(); + cached_response(&stale.response, "STALE") + } + + /// the stored body with the fields of the origin's 304, which also + /// freshens the stored response. + fn revalidated(&self, stale: &StaleEntry, head: &Head) -> HttpResponse { + let fresh = self.state.cache.freshen( + stale, + &self.request_policy, + self.req.headers(), + &stored_fields(forwardable_fields(head)), + head.map.get(header::AGE), + ); + self.state.cache_stats.revalidation(); + cached_response(&fresh, "REVALIDATED") + } +} + +/// `builder` with the conditional fields from `stored`'s validators (RFC +/// 9111 section 4.3.1). +fn with_validators( + mut builder: hyper::http::request::Builder, + stored: &StoredResponse, +) -> hyper::http::request::Builder { + for (name, value) in stored.validators() { + builder = builder.header(name.as_str(), value.as_bytes()); + } + builder +} + +/// the origin request from `builder` and the client's body, or the status +/// of the client's error response. +async fn origin_request( + builder: hyper::http::request::Builder, + payload: web::Payload, +) -> Result, StatusCode> { + let body = request_body(payload).await.map_err(|e| { + warn!("Failed to read request body: {}", e); + StatusCode::BAD_REQUEST + })?; + builder.body(body).map_err(|e| { + error!("Failed to build hyper request: {}", e); + StatusCode::INTERNAL_SERVER_ERROR + }) +} + +/// a plain error response, such as "bad gateway" for a 502. +fn error_response(status: StatusCode) -> HttpResponse { + let reason = status.canonical_reason().unwrap_or_default(); + HttpResponse::build(status).body(reason.to_ascii_lowercase()) +} + +fn empty_body() -> UpstreamBody { + Empty::new().map_err(|never| match never {}).boxed_unsync() } /// the origin request for `req`, without its body, and its target URI. @@ -200,7 +457,7 @@ fn store_plan( Some(Storing { limit: state.cache.max_entry(), finish: Box::new(move |status, headers, body| { - cache.insert_response(cache_key, vary, &storable, status, headers, body) + cache.insert_response(cache_key, vary, &storable, status, headers, body); }), }) } @@ -212,7 +469,7 @@ async fn request_body(mut payload: web::Payload) -> Result return Ok(Empty::new().map_err(|never| match never {}).boxed_unsync()), + None => return Ok(empty_body()), Some(chunk) => chunk?, }; @@ -303,19 +560,12 @@ struct BodyCopy { fn client_response(head: Head, body: Incoming, store: Option) -> HttpResponse { let mut builder = HttpResponse::build(head.status); - let options = connection_options(head.map.get_all(header::CONNECTION)); - let mut stored_headers = Vec::new(); - for (name, value) in &head.fields { - // actix writes Content-Length from the body size, so a copied header - // would go stale when the Compress middleware re-encodes the body - if name != header::CONTENT_LENGTH && name != CACHE_STATUS && is_end_to_end(name, &options) { - builder.append_header((name.clone(), value.clone())); - if name != header::AGE { - stored_headers.push((name.clone(), value.clone())); - } - } + let fields = forwardable_fields(&head); + for (name, value) in &fields { + builder.append_header((name.clone(), value.clone())); } builder.insert_header((CACHE_STATUS, "MISS")); + let stored_headers = stored_fields(fields); // the data stream drops trailer frames let body = body.into_data_stream().map(|chunk| { @@ -347,16 +597,37 @@ fn client_response(head: Head, body: Incoming, store: Option) -> HttpRe } } -/// the client's response from a stored one (RFC 9111 section 4). actix -/// leaves the body out when it answers a HEAD request. -fn stored_response(stored: &StoredResponse) -> HttpResponse { +/// the origin response's fields that go to the client. +fn forwardable_fields(head: &Head) -> Fields { + let options = connection_options(head.map.get_all(header::CONNECTION)); + head.fields + .iter() + // actix writes Content-Length from the body size, so a copied header + // would go stale when the Compress middleware re-encodes the body + .filter(|(name, _)| { + name != header::CONTENT_LENGTH && name != CACHE_STATUS && is_end_to_end(name, &options) + }) + .cloned() + .collect() +} + +/// forwardable fields without `Age`, which the store works out on each use. +fn stored_fields(mut fields: Fields) -> Fields { + fields.retain(|(name, _)| name != header::AGE); + fields +} + +/// the client's response from a stored one (RFC 9111 section 4), marked +/// with `cache_status`. actix leaves the body out when it answers a HEAD +/// request. +fn cached_response(stored: &StoredResponse, cache_status: &'static str) -> HttpResponse { let mut builder = HttpResponse::build(stored.status); for (name, value) in &stored.headers { builder.append_header((name.clone(), value.clone())); } builder .insert_header((header::AGE, stored.age().as_secs().to_string())) - .insert_header((CACHE_STATUS, "HIT")) + .insert_header((CACHE_STATUS, cache_status)) .body(stored.body.clone()) } diff --git a/tests/integration/main.rs b/tests/integration/main.rs index fd848c2..9552396 100644 --- a/tests/integration/main.rs +++ b/tests/integration/main.rs @@ -4,5 +4,6 @@ mod common; mod forwarded; mod http2; mod proxy; +mod revalidation; mod smoke; mod tls; diff --git a/tests/integration/revalidation.rs b/tests/integration/revalidation.rs new file mode 100644 index 0000000..4bdd073 --- /dev/null +++ b/tests/integration/revalidation.rs @@ -0,0 +1,514 @@ +use crate::common::{self, cache_status}; + +use actix_web::http::StatusCode; +use actix_web::test; +use std::io::{BufRead, BufReader, Write}; +use std::net::TcpListener; +use std::time::{Duration, Instant}; +use wiremock::matchers::{any, header}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const KIB: usize = 1024; + +/// how long a `max-age=1` response takes to go stale, with a margin. +const PAST_ONE_SECOND: Duration = Duration::from_millis(1100); + +fn get(uri: &str) -> test::TestRequest { + test::TestRequest::get().uri(uri) +} + +/// a 200 with `cache_control`, an `ETag` of `"v1"` and the body "old". +fn tagged(cache_control: &str) -> ResponseTemplate { + ResponseTemplate::new(200) + .insert_header("cache-control", cache_control) + .insert_header("etag", "\"v1\"") + .set_body_string("old") +} + +/// an origin that answers the first request with `first` and every later +/// request with `later`. +async fn origin_answering(first: ResponseTemplate, later: ResponseTemplate) -> MockServer { + let origin = MockServer::start().await; + Mock::given(any()) + .respond_with(first) + .up_to_n_times(1) + .mount(&origin) + .await; + Mock::given(any()).respond_with(later).mount(&origin).await; + origin +} + +async fn origin_requests(origin: &MockServer) -> usize { + origin.received_requests().await.unwrap().len() +} + +/// the value of `name` on the origin's `n`th received request. +async fn received_header(origin: &MockServer, n: usize, name: &str) -> Option { + let received = origin.received_requests().await.unwrap(); + received[n] + .headers + .get(name) + .map(|v| v.to_str().unwrap().to_owned()) +} + +async fn health(app: &S) -> serde_json::Value +where + S: actix_web::dev::Service< + actix_http::Request, + Response = actix_web::dev::ServiceResponse, + Error = actix_web::Error, + >, + B: actix_web::body::MessageBody, +{ + let resp = test::call_service(app, get("/health").to_request()).await; + test::read_body_json(resp).await +} + +fn age_of(resp: &actix_web::dev::ServiceResponse) -> u64 { + resp.headers() + .get("age") + .unwrap() + .to_str() + .unwrap() + .parse() + .unwrap() +} + +/// an origin that answers one connection with `response` and then stops +/// listening, so later requests are refused. +fn one_shot_origin(response: String) -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + drop(listener); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut line = String::new(); + while reader.read_line(&mut line).unwrap_or(0) > 0 && line != "\r\n" { + line.clear(); + } + let mut stream = stream; + let _ = stream.write_all(response.as_bytes()); + }); + format!("http://{addr}") +} + +/// a 200 with `cache_control`, an `ETag` of `"v1"` and the body "old", that +/// closes its connection. +fn one_shot_response(cache_control: &str) -> String { + format!( + "HTTP/1.1 200 OK\r\ncache-control: {cache_control}\r\netag: \"v1\"\r\nconnection: close\r\ncontent-length: 3\r\n\r\nold" + ) +} + +#[actix_web::test] +async fn not_modified_serves_the_stored_body_with_updated_headers() { + let origin = origin_answering( + tagged("max-age=1").insert_header("x-version", "1"), + ResponseTemplate::new(304) + .insert_header("cache-control", "max-age=60") + .insert_header("etag", "\"v1\"") + .insert_header("x-version", "2"), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let revalidated = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(revalidated.status(), StatusCode::OK); + assert_eq!(cache_status(&revalidated), "REVALIDATED"); + assert_eq!(revalidated.headers().get("x-version").unwrap(), "2"); + assert_eq!( + revalidated.headers().get("cache-control").unwrap(), + "max-age=60" + ); + assert_eq!(test::read_body(revalidated).await.as_ref(), b"old"); + assert_eq!( + received_header(&origin, 1, "if-none-match") + .await + .as_deref(), + Some("\"v1\"") + ); + + // the 304's max-age=60 made the entry fresh again + let after = test::call_service(&app, get("/page").to_request()).await; + assert_eq!(cache_status(&after), "HIT"); + assert_eq!(after.headers().get("x-version").unwrap(), "2"); + assert_eq!(test::read_body(after).await.as_ref(), b"old"); + assert_eq!(origin_requests(&origin).await, 2); + + let health = health(&app).await; + assert_eq!(health["cache"]["revalidations"], 1); +} + +#[actix_web::test] +async fn last_modified_is_revalidated_with_if_modified_since() { + let last_modified = "Tue, 01 Sep 2026 10:00:00 GMT"; + let origin = MockServer::start().await; + // `header` splits values on commas, which an HTTP date contains + let if_modified_since = move |request: &wiremock::Request| { + request + .headers + .get("if-modified-since") + .map(|v| v.as_bytes()) + == Some(last_modified.as_bytes()) + }; + Mock::given(if_modified_since) + .respond_with(ResponseTemplate::new(304).insert_header("cache-control", "max-age=60")) + .mount(&origin) + .await; + Mock::given(any()) + .respond_with( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=1") + .insert_header("last-modified", last_modified) + .set_body_string("old"), + ) + .mount(&origin) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let revalidated = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(revalidated.status(), StatusCode::OK); + assert_eq!(cache_status(&revalidated), "REVALIDATED"); + assert_eq!(test::read_body(revalidated).await.as_ref(), b"old"); + assert_eq!(received_header(&origin, 1, "if-none-match").await, None); + let after = test::call_service(&app, get("/page").to_request()).await; + assert_eq!(cache_status(&after), "HIT"); +} + +#[actix_web::test] +async fn full_response_to_revalidation_replaces_the_entry() { + let origin = origin_answering( + tagged("max-age=1"), + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .insert_header("etag", "\"v2\"") + .set_body_string("new"), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let replaced = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(cache_status(&replaced), "MISS"); + assert_eq!(test::read_body(replaced).await.as_ref(), b"new"); + assert_eq!( + received_header(&origin, 1, "if-none-match") + .await + .as_deref(), + Some("\"v1\"") + ); + let after = test::call_service(&app, get("/page").to_request()).await; + assert_eq!(cache_status(&after), "HIT"); + assert_eq!(test::read_body(after).await.as_ref(), b"new"); + assert_eq!(origin_requests(&origin).await, 2); +} + +#[actix_web::test] +async fn client_conditional_passes_through_and_keeps_the_entry() { + let origin = MockServer::start().await; + Mock::given(header("if-none-match", "\"client\"")) + .respond_with(ResponseTemplate::new(304).insert_header("etag", "\"client\"")) + .mount(&origin) + .await; + Mock::given(header("if-none-match", "\"v1\"")) + .respond_with(ResponseTemplate::new(304).insert_header("cache-control", "max-age=60")) + .mount(&origin) + .await; + Mock::given(any()) + .respond_with(tagged("max-age=1")) + .mount(&origin) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let conditional = get("/page").insert_header(("if-none-match", "\"client\"")); + let passed = test::call_service(&app, conditional.to_request()).await; + + assert_eq!(passed.status(), StatusCode::NOT_MODIFIED); + assert_eq!(cache_status(&passed), "MISS"); + assert_eq!( + received_header(&origin, 1, "if-none-match") + .await + .as_deref(), + Some("\"client\"") + ); + + // the client's 304 left the stored entry in place for revalidation + let plain = test::call_service(&app, get("/page").to_request()).await; + assert_eq!(cache_status(&plain), "REVALIDATED"); + assert_eq!(test::read_body(plain).await.as_ref(), b"old"); +} + +#[actix_web::test] +async fn stale_while_revalidate_serves_stale_and_refreshes_in_the_background() { + let origin = origin_answering( + tagged("max-age=1, stale-while-revalidate=10"), + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=60") + .set_body_string("new") + .set_delay(Duration::from_millis(500)), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let started = Instant::now(); + let stale = test::call_service(&app, get("/page").to_request()).await; + + assert!( + started.elapsed() < Duration::from_millis(400), + "waited for the origin" + ); + assert_eq!(stale.status(), StatusCode::OK); + assert_eq!(cache_status(&stale), "STALE"); + assert_eq!(test::read_body(stale).await.as_ref(), b"old"); + + let refreshed = wait_for_hit(&app, "/page").await; + assert_eq!(refreshed.as_ref(), b"new"); + assert_eq!(origin_requests(&origin).await, 2); + assert_eq!( + received_header(&origin, 1, "if-none-match") + .await + .as_deref(), + Some("\"v1\"") + ); + + // polls before the refresh lands are stale serves too + let health = health(&app).await; + assert!(health["cache"]["stale"].as_u64().unwrap() >= 1); + assert_eq!(health["cache"]["background_refreshes"], 1); +} + +/// requests `path` until it is a hit, for up to three seconds, and returns +/// the hit's body. +async fn wait_for_hit(app: &S, path: &str) -> bytes::Bytes +where + S: actix_web::dev::Service< + actix_http::Request, + Response = actix_web::dev::ServiceResponse, + Error = actix_web::Error, + >, + B: actix_web::body::MessageBody, +{ + let deadline = Instant::now() + Duration::from_secs(3); + loop { + actix_web::rt::time::sleep(Duration::from_millis(100)).await; + let resp = test::call_service(app, get(path).to_request()).await; + let status = cache_status(&resp); + let body = test::read_body(resp).await; + if status == "HIT" { + return body; + } + assert!(Instant::now() < deadline, "no hit, last status {status}"); + } +} + +#[actix_web::test] +async fn concurrent_stale_requests_start_one_background_revalidation() { + let origin = origin_answering( + tagged("max-age=1, stale-while-revalidate=10"), + ResponseTemplate::new(304) + .insert_header("cache-control", "max-age=60") + .set_delay(Duration::from_millis(500)), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let requests = (0..5).map(|_| test::call_service(&app, get("/page").to_request())); + let responses = futures_util::future::join_all(requests).await; + for resp in responses { + assert_eq!(cache_status(&resp), "STALE"); + assert_eq!(test::read_body(resp).await.as_ref(), b"old"); + } + + assert_eq!(wait_for_hit(&app, "/page").await.as_ref(), b"old"); + assert_eq!(origin_requests(&origin).await, 2); + let health = health(&app).await; + assert!(health["cache"]["stale"].as_u64().unwrap() >= 5); + assert_eq!(health["cache"]["background_refreshes"], 1); + assert_eq!(health["cache"]["revalidations"], 1); +} + +#[actix_web::test] +async fn stale_carries_the_right_age() { + let origin = origin_answering( + tagged("max-age=6, stale-while-revalidate=10").insert_header("age", "5"), + ResponseTemplate::new(304).set_delay(Duration::from_secs(5)), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let stale = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(cache_status(&stale), "STALE"); + let age = age_of(&stale); + assert!((6..=7).contains(&age), "age was {age}"); +} + +#[actix_web::test] +async fn stale_if_error_serves_stale_on_origin_errors() { + for status in [500, 502, 503, 504] { + let origin = origin_answering( + tagged("max-age=1, stale-if-error=10"), + ResponseTemplate::new(status), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let stale = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(stale.status(), StatusCode::OK, "origin {status}"); + assert_eq!(cache_status(&stale), "STALE", "origin {status}"); + assert_eq!(test::read_body(stale).await.as_ref(), b"old"); + } +} + +#[actix_web::test] +async fn stale_if_error_serves_stale_on_timeout() { + let origin = origin_answering( + tagged("max-age=1, stale-if-error=10"), + ResponseTemplate::new(200).set_delay(Duration::from_secs(3)), + ) + .await; + let (app, _assets) = + common::service_with(&origin.uri(), |c| c.upstream_timeout_seconds = 1).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let stale = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(stale.status(), StatusCode::OK); + assert_eq!(cache_status(&stale), "STALE"); + assert_eq!(test::read_body(stale).await.as_ref(), b"old"); +} + +#[actix_web::test] +async fn stale_if_error_serves_stale_on_connection_refused() { + let origin = one_shot_origin(one_shot_response("max-age=1, stale-if-error=10")); + let (app, _assets) = common::service(&origin).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let stale = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(stale.status(), StatusCode::OK); + assert_eq!(cache_status(&stale), "STALE"); + assert_eq!(test::read_body(stale).await.as_ref(), b"old"); +} + +#[actix_web::test] +async fn origin_error_goes_through_outside_the_stale_if_error_window() { + let origin = origin_answering( + tagged("max-age=1, stale-if-error=1"), + ResponseTemplate::new(503), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(Duration::from_millis(2200)).await; + let resp = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(cache_status(&resp), "MISS"); +} + +#[actix_web::test] +async fn must_revalidate_is_never_served_stale() { + for directive in ["must-revalidate", "proxy-revalidate"] { + let cache_control = + format!("max-age=1, {directive}, stale-while-revalidate=10, stale-if-error=10"); + let origin = origin_answering(tagged(&cache_control), ResponseTemplate::new(503)).await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let resp = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!( + resp.status(), + StatusCode::SERVICE_UNAVAILABLE, + "{directive}" + ); + assert_eq!(cache_status(&resp), "MISS", "{directive}"); + assert_eq!(origin_requests(&origin).await, 2, "{directive}"); + } +} + +#[actix_web::test] +async fn must_revalidate_gets_504_when_the_origin_is_unreachable() { + for directive in ["must-revalidate", "proxy-revalidate"] { + let cache_control = format!("max-age=1, {directive}, stale-if-error=10"); + let origin = one_shot_origin(one_shot_response(&cache_control)); + let (app, _assets) = common::service(&origin).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let resp = test::call_service(&app, get("/page").to_request()).await; + + assert_eq!(resp.status(), StatusCode::GATEWAY_TIMEOUT, "{directive}"); + } +} + +#[actix_web::test] +async fn entry_without_validator_or_stale_window_is_gone_after_expiry() { + let origin = origin_answering( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=1") + .set_body_string("old"), + ResponseTemplate::new(200).set_body_string("new"), + ) + .await; + let (app, _assets) = common::service(&origin.uri()).await; + + test::read_body(test::call_service(&app, get("/page").to_request()).await).await; + assert_eq!(health(&app).await["cache"]["items"], 1); + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + + assert_eq!(health(&app).await["cache"]["items"], 0); + let resp = test::call_service(&app, get("/page").to_request()).await; + assert_eq!(cache_status(&resp), "MISS"); + assert_eq!(test::read_body(resp).await.as_ref(), b"new"); +} + +#[actix_web::test] +async fn stale_entries_kept_for_revalidation_stay_within_the_byte_bound() { + let origin = MockServer::start().await; + Mock::given(any()) + .respond_with( + ResponseTemplate::new(200) + .insert_header("cache-control", "max-age=1, stale-while-revalidate=60") + .insert_header("etag", "\"v1\"") + .set_body_bytes(vec![b'x'; 300 * KIB]), + ) + .mount(&origin) + .await; + let (app, _assets) = common::service_with(&origin.uri(), |c| c.cache_size_mb = 1).await; + + for path in ["/a", "/b", "/c", "/d", "/e"] { + test::read_body(test::call_service(&app, get(path).to_request()).await).await; + } + actix_web::rt::time::sleep(PAST_ONE_SECOND).await; + let health = health(&app).await; + + let items = health["cache"]["items"].as_u64().unwrap(); + assert!((1..=3).contains(&items), "{items} items"); + assert!(health["cache"]["bytes"].as_u64().unwrap() <= 1024 * KIB as u64); +} From 3377b9ffab22867f44fc59a6371ee519813aa9f8 Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:01:19 +0100 Subject: [PATCH 2/3] docs(readme): describe revalidation and stale responses --- README.md | 42 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 36 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 0cb576c..6400c76 100644 --- a/README.md +++ b/README.md @@ -66,17 +66,46 @@ For each request: - A request with `X-HTTP-Method-Override`, `X-HTTP-Method` or `X-Method-Override` is forwarded with that header, but is neither served from the cache nor stored, because the origin may treat it as another method. - A request with an unsafe method, such as `POST`, `PUT` or `DELETE`, that gets a 2xx or 3xx response removes the stored response for its URL. -Responses served from the cache carry `Age`. Proxied and asset responses carry `X-Shadowstep-Cache: HIT` or `MISS`. +Responses served from the cache carry `Age`. Proxied and asset responses carry `X-Shadowstep-Cache`: -`--cache-ttl-seconds` (default 300) caps how long any entry is kept, whatever the origin's freshness lifetime. `--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. +- `HIT`: a fresh stored response. +- `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`. + +`--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. + +### Revalidation and stale responses + +A stored response that is stale stays in the cache for a grace period so that it can be revalidated or served stale. The grace period is `--cache-ttl-seconds` if the response has an `ETag` or `Last-Modified`, and otherwise the larger of its `stale-while-revalidate` and `stale-if-error` windows, capped at `--cache-ttl-seconds`. Those windows do not count for a response that must be revalidated, as described below. A response with neither a validator nor a stale window is dropped when it goes stale. Entries in their grace period count towards `--cache-size-mb`. + +A `GET` that finds a stale response sends the origin `If-None-Match` from the stored `ETag` and `If-Modified-Since` from the stored `Last-Modified`. A request whose `Cache-Control: max-age` is shorter than a fresh stored response's age revalidates that response the same way. + +- On `304 Not Modified`, the 304's header fields replace the stored fields of the same name, the freshness lifetime starts again from the updated fields, and the client gets `200` with the stored body. +- On any other response, the proxy forwards it and stores it under the usual rules. +- If the client sent its own `If-None-Match`, `If-Modified-Since`, `If-Match`, `If-Unmodified-Since` or `If-Range`, the proxy forwards those unchanged and adds none of its own. The origin's answer, including a `304`, goes to the client as `MISS`, and a 304 leaves the stored response as it was. + +`Cache-Control: stale-while-revalidate=N` lets the proxy serve a response for `N` seconds after it goes stale while it revalidates in the background. At most one background revalidation runs for each stored response at a time. It uses the same header fields, forwarding headers and `--upstream-timeout-seconds` as any request to the origin, with the stored validators in place of the client's conditional headers. + +`Cache-Control: stale-if-error=N` lets the proxy serve a response for `N` seconds after it goes stale when the origin answers 500, 502, 503 or 504, refuses the connection or times out. + +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. Assets share the same cache. A stored asset is read from disk again when the file's size or modified time changes. -`/health` counts hits and misses for proxied requests and assets together. `items` and `bytes` describe the whole cache. +`/health` counts responses for proxied requests and assets together. Each response counts once: + +- `hits`: fresh stored responses. +- `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`. + +`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. Known limits: -- There is no revalidation. shadowstep never sends conditional requests to the origin, so a stale entry is dropped and fetched again in full. +- 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. - 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. @@ -137,7 +166,7 @@ Each option can be set with a flag or an environment variable. The flag wins if | `--origin-url` | `ORIGIN_URL` | required | Upstream origin URL, for example `http://origin.internal:3000` | | `--listen-addr` | `LISTEN_ADDR` | `0.0.0.0:8080` | Address for the plain HTTP listener | | `--asset-path` | `ASSET_PATH` | `/app/assets` | Directory served under `/assets/` | -| `--cache-ttl-seconds` | `CACHE_TTL_SECONDS` | `300` | Longest time any cache entry is kept; 0 turns caching off | +| `--cache-ttl-seconds` | `CACHE_TTL_SECONDS` | `300` | Longest freshness lifetime of any cache entry, and longest time a stale one is kept; 0 turns caching off | | `--cache-size-mb` | `CACHE_SIZE_MB` | `100` | Cache size in MiB for origin responses and assets together; 0 turns caching off | | `--tls-cert` | `TLS_CERT_PATH` | none | PEM certificate chain | | `--tls-key` | `TLS_KEY_PATH` | none | PEM private key in PKCS#8 form | @@ -226,13 +255,14 @@ The Deployment runs two replicas, and each has its own cache. `X-Forwarded-For` cargo test ``` -Unit tests in `src/assets.rs` cover asset path traversal, `src/cache.rs` covers `Cache-Control` parsing, and `src/forwarded.rs` covers reading the client's address, scheme and host from the connection. The integration tests in `tests/integration/` build one test crate and run the app against a [wiremock](https://crates.io/crates/wiremock) origin or a local TCP origin: +Unit tests in `src/assets.rs` cover asset path traversal, `src/cache.rs` covers `Cache-Control` parsing and header updates from a 304, and `src/forwarded.rs` covers reading the client's address, scheme and host from the connection. The integration tests in `tests/integration/` build one test crate and run the app against a [wiremock](https://crates.io/crates/wiremock) origin or a local TCP origin: - `smoke.rs`: `/health`, proxying, upstream paths and the 502 path - `proxy.rs`: request and response bodies, the upstream timeout, hop-by-hop and URL override headers - `tls.rs`: the HTTPS listener and `--tls-listen-addr` - `forwarded.rs`: removal and replacement of forwarding headers - `cache.rs`: origin response caching +- `revalidation.rs`: conditional requests, `stale-while-revalidate`, `stale-if-error` and `must-revalidate` CI also runs: From 06c16967d24c21bb221e05a63da6731dbd230d5f Mon Sep 17 00:00:00 2001 From: jamie Date: Sun, 27 Sep 2026 17:14:16 +0100 Subject: [PATCH 3/3] refactor(cache): use parking_lot for the refresh set --- Cargo.lock | 1 + Cargo.toml | 1 + src/cache.rs | 15 +++++---------- 3 files changed, 7 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bba96af..b02e298 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1938,6 +1938,7 @@ dependencies = [ "log", "mime_guess", "moka", + "parking_lot", "rcgen", "rustls", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 94f1a23..124b0c6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ moka = { version = "0.12", features = ["sync"] } bytes = "1" httpdate = "1.0.3" futures-util = { version = "0.3", default-features = false, features = ["std"] } +parking_lot = "0.12" [profile.release] lto = true diff --git a/src/cache.rs b/src/cache.rs index eed6f14..770c64b 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -6,10 +6,11 @@ use actix_web::http::{Method, StatusCode}; use bytes::Bytes; use moka::sync::Cache; use moka::Expiry; +use parking_lot::Mutex; use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; const MIB: u64 = 1024 * 1024; @@ -225,17 +226,10 @@ pub struct RefreshGuard { impl Drop for RefreshGuard { fn drop(&mut self) { - lock(&self.refreshing).remove(&self.key); + self.refreshing.lock().remove(&self.key); } } -/// the set of keys being refreshed. a panic while the lock is held cannot -/// leave the set inconsistent, so a poisoned lock is still usable. -fn lock(set: &Mutex>) -> std::sync::MutexGuard<'_, HashSet> { - set.lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) -} - fn entry_weight(key: &Key, entry: &Entry) -> u32 { let key = match key { Key::Asset(path) => path.as_os_str().len(), @@ -395,7 +389,8 @@ impl Store { /// claims the background revalidation of `entry`, or `None` when one is /// already running. pub fn start_refresh(&self, entry: &StaleEntry) -> Option { - lock(&self.refreshing) + self.refreshing + .lock() .insert(entry.key.clone()) .then(|| RefreshGuard { refreshing: self.refreshing.clone(),