rust_web_server/cache/

mod.rs

//! In-memory response cache middleware.
//!
//! [`CacheLayer`] implements [`Middleware`] and short-circuits the inner
//! application for cacheable `GET` responses within their TTL.  Entries are
//! bounded by a configurable capacity; the oldest entry is evicted when the
//! store is full and no expired entries remain.
//!
//! # What is cached
//!
//! - Method: **GET only**; all other methods bypass the cache.
//! - Status: 2xx responses (200, 201, 203, 204, 206, …).
//! - Response `Cache-Control: no-store` or `private` — **not** cached.
//! - Private or partial requests — `Authorization`, `Cookie` or `Range` — skip
//!   the cache entirely (no lookup, no store): their responses may differ per
//!   user or cover only part of the resource.
//! - Responses that set a cookie (`Set-Cookie`) or stream their body
//!   (`stream_file` / `stream_pipe`, not held in memory) are **not** cached.
//! - Request `Cache-Control: no-cache` — cache is bypassed, handler is called,
//!   but the fresh response **is** stored (revalidation).
//! - How long: the response's `Cache-Control: s-maxage`, else `max-age`,
//!   capped by the layer's [`ttl`](CacheLayer::ttl); the `ttl` when it says
//!   neither. `max-age=0` is not stored (since 18.11.0; before, every
//!   response was kept for the `ttl`). Hits carry an `Age` header.
//!
//! # Example
//!
//! ```rust,no_run
//! use rust_web_server::app::App;
//! use rust_web_server::cache::CacheLayer;
//! use rust_web_server::core::New;
//!
//! let app = App::new()
//!     .wrap(CacheLayer::memory(1000).ttl(60).vary_by_header("Accept"));
//! ```

#[cfg(test)]
mod tests;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) mod disk;

use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use crate::application::Application;
use crate::header::Header;
use crate::middleware::Middleware;
use crate::request::{METHOD, Request};
use crate::response::Response;
use crate::server::ConnectionInfo;

// ── cache store ───────────────────────────────────────────────────────────────

struct CachedEntry {
    response: Response,
    inserted_at: Instant,
    /// How long this entry is fresh: the layer's `ttl`, or less when the
    /// response's `Cache-Control` said so.
    ttl: Duration,
}

struct CacheStore {
    entries: HashMap<String, CachedEntry>,
    /// Insertion order — front is oldest; used for capacity eviction.
    order: VecDeque<String>,
}

impl CacheStore {
    fn new() -> Self {
        CacheStore { entries: HashMap::new(), order: VecDeque::new() }
    }

    fn get(&self, key: &str) -> Option<&CachedEntry> {
        self.entries.get(key).filter(|e| e.inserted_at.elapsed() < e.ttl)
    }

    fn insert(&mut self, key: String, entry: CachedEntry, capacity: usize) {
        // Update in place without disturbing insertion order.
        if self.entries.contains_key(&key) {
            self.entries.insert(key, entry);
            return;
        }
        // `purge_expired` is called by the caller before `insert`, so any
        // remaining entries are still live. Evict the oldest if we're full.
        if self.entries.len() >= capacity {
            if let Some(oldest) = self.order.pop_front() {
                self.entries.remove(&oldest);
            }
        }
        self.order.push_back(key.clone());
        self.entries.insert(key, entry);
    }

    fn purge_expired(&mut self) {
        let expired: Vec<String> = self.entries.iter()
            .filter(|(_, e)| e.inserted_at.elapsed() >= e.ttl)
            .map(|(k, _)| k.clone())
            .collect();
        for k in &expired {
            self.entries.remove(k);
            self.order.retain(|o| o != k);
        }
    }
}

/// How long `response` may be served from a shared cache: its
/// `Cache-Control: s-maxage`, else `max-age` (RFC 9111 §5.2.2), capped by
/// `cap`, the layer's `ttl`; `cap` when it says neither. Zero: don't store
/// (`max-age=0`). A directive whose value isn't a number is ignored.
pub(crate) fn freshness(response: &Response, cap: Duration) -> Duration {
    let mut s_maxage = None;
    let mut max_age = None;
    for h in response.headers.iter().filter(|h| h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL)) {
        for directive in h.value.split(',') {
            let Some((name, value)) = directive.split_once('=') else { continue };
            let Ok(secs) = value.trim().trim_matches('"').parse::<u64>() else { continue };
            match name.trim().to_ascii_lowercase().as_str() {
                "s-maxage" => s_maxage = Some(s_maxage.map_or(secs, |v: u64| v.min(secs))),
                "max-age" => max_age = Some(max_age.map_or(secs, |v: u64| v.min(secs))),
                _ => {}
            }
        }
    }
    match s_maxage.or(max_age) {
        Some(secs) => cap.min(Duration::from_secs(secs)),
        None => cap,
    }
}

// ── purge ─────────────────────────────────────────────────────────────────────

/// What [`CacheLayer::purge`] removes. A cache key is the request target
/// (path and query) plus the values of `vary_by_header`s, so `Exact` removes
/// every variant of one target.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Purge {
    /// Everything.
    All,
    /// One request target, e.g. `/products/42` or `/search?q=lamp`.
    Exact(String),
    /// Every target starting with this, e.g. `/products/`.
    Prefix(String),
}

impl Purge {
    fn matches(&self, key: &str) -> bool {
        let target = key.split('\x00').next().unwrap_or(key);
        match self {
            Purge::All => true,
            Purge::Exact(t) => target == t,
            Purge::Prefix(p) => target.starts_with(p.as_str()),
        }
    }

    /// A `PURGE` request's target: a trailing `*` makes it a prefix.
    pub fn from_target(target: &str) -> Purge {
        match target.strip_suffix('*') {
            Some(prefix) => Purge::Prefix(prefix.to_string()),
            None => Purge::Exact(target.to_string()),
        }
    }
}

/// Equal, in time that depends only on the lengths.
fn same_bytes(a: &[u8], b: &[u8]) -> bool {
    a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}

/// Named caches for the admin API: the binary's own and one per
/// config-file route that has `[route.middleware.cache]`.
pub mod registry {
    use super::CacheLayer;
    use std::collections::BTreeMap;
    use std::sync::{Mutex, OnceLock};

    fn caches() -> &'static Mutex<BTreeMap<String, CacheLayer>> {
        static CACHES: OnceLock<Mutex<BTreeMap<String, CacheLayer>>> = OnceLock::new();
        CACHES.get_or_init(Mutex::default)
    }

    /// Make `cache` reachable as `name` (a clone: it shares the store).
    pub fn register(name: impl Into<String>, cache: CacheLayer) {
        caches().lock().unwrap_or_else(|e| e.into_inner()).insert(name.into(), cache);
    }

    /// Replace every cache whose name starts with `group` by `caches` (a new
    /// config generation's routes).
    pub(crate) fn replace_group(group: &str, new: Vec<(String, CacheLayer)>) {
        let mut map = caches().lock().unwrap_or_else(|e| e.into_inner());
        map.retain(|name, _| !name.starts_with(group));
        map.extend(new);
    }

    /// Every registered cache, by name.
    pub fn list() -> Vec<(String, CacheLayer)> {
        caches().lock().unwrap_or_else(|e| e.into_inner()).iter().map(|(n, c)| (n.clone(), c.clone())).collect()
    }

    pub fn get(name: &str) -> Option<CacheLayer> {
        caches().lock().unwrap_or_else(|e| e.into_inner()).get(name).cloned()
    }
}

// ── CacheLayer ────────────────────────────────────────────────────────────────

/// An in-memory response cache middleware.
///
/// Construct with [`CacheLayer::memory`] and configure with the builder methods
/// [`ttl`](CacheLayer::ttl) and [`vary_by_header`](CacheLayer::vary_by_header).
///
/// `CacheLayer` is cheaply `Clone`-able — clones share the same underlying
/// store, which lets you keep a handle for cache stats and invalidation while
/// the other clone is used as middleware.
#[derive(Clone)]
pub struct CacheLayer {
    store: Arc<Mutex<CacheStore>>,
    /// `store`'s entry count, kept after every change under its lock: an
    /// empty cache (nothing cacheable seen yet — e.g. static files, sent
    /// `no-cache`) costs a request no key and no lock. The lock was shared
    /// by every worker and taken on every GET.
    stored: Arc<AtomicUsize>,
    hits: Arc<AtomicU64>,
    misses: Arc<AtomicU64>,
    capacity: usize,
    ttl: Duration,
    vary_headers: Vec<String>,
    /// [`CacheLayer::disk`]: entries in files instead of `store`.
    #[cfg(not(target_arch = "wasm32"))]
    disk: Option<Arc<disk::DiskStore>>,
    /// [`CacheLayer::purge_token`]: accept `PURGE` with this bearer token.
    purge_token: Option<Arc<String>>,
}

impl CacheLayer {
    /// Create a new in-memory cache bounded to `capacity` entries.
    ///
    /// Default TTL is **60 seconds**. Adjust with [`.ttl()`](CacheLayer::ttl).
    pub fn memory(capacity: usize) -> Self {
        CacheLayer {
            store: Arc::new(Mutex::new(CacheStore::new())),
            stored: Arc::new(AtomicUsize::new(0)),
            hits: Arc::new(AtomicU64::new(0)),
            misses: Arc::new(AtomicU64::new(0)),
            capacity,
            ttl: Duration::from_secs(60),
            vary_headers: vec![],
            #[cfg(not(target_arch = "wasm32"))]
            disk: None,
            purge_token: None,
        }
    }

    /// Accept the HTTP `PURGE` method on the routes behind this layer,
    /// authenticated by `Authorization: Bearer <token>` (since 18.11.0).
    /// `PURGE /products/42` removes that target (every `Vary` variant);
    /// `PURGE /products/*` removes the prefix. The answer is
    /// `{"purged":<n>}`; a missing or wrong token gets `401`. Without a
    /// token, `PURGE` passes through to the application as before.
    pub fn purge_token(mut self, token: impl Into<String>) -> Self {
        let token = token.into();
        self.purge_token = (!token.is_empty()).then(|| Arc::new(token));
        self
    }

    /// Remove the entries `target` names; how many.
    pub fn purge(&self, target: &Purge) -> usize {
        if let Some(disk) = self.disk_store() {
            let n = disk.remove_where(|key| target.matches(key));
            self.stored.store(disk.len(), Ordering::Relaxed);
            return n;
        }
        let mut guard = self.store.lock().unwrap_or_else(|e| e.into_inner());
        let keys: Vec<String> = guard.entries.keys().filter(|k| target.matches(k)).cloned().collect();
        for key in &keys {
            guard.entries.remove(key);
        }
        guard.order.retain(|k| !keys.contains(k));
        self.stored.store(guard.entries.len(), Ordering::Relaxed);
        keys.len()
    }

    /// `"memory"` or `"disk"`.
    pub fn kind(&self) -> &'static str {
        if self.disk_store().is_some() { "disk" } else { "memory" }
    }

    /// The `PURGE` answer, when this request is one and a token is set.
    fn answer_purge(&self, request: &Request) -> Option<Response> {
        let token = self.purge_token.as_ref()?;
        if !request.method.eq_ignore_ascii_case("PURGE") {
            return None;
        }
        use crate::response::STATUS_CODE_REASON_PHRASE;
        let sent = request.header(Header::_AUTHORIZATION).and_then(|h| h.value.strip_prefix("Bearer ").map(str::trim));
        let authorized = sent.is_some_and(|s| same_bytes(s.as_bytes(), token.as_bytes()));
        if !authorized {
            let mut response = Response::json(STATUS_CODE_REASON_PHRASE.n401_unauthorized, br#"{"error":"PURGE needs the purge token"}"#.to_vec());
            response.headers.push(Header::new("WWW-Authenticate", r#"Bearer realm="cache purge""#));
            return Some(response);
        }
        let n = self.purge(&Purge::from_target(&request.request_uri));
        Some(Response::json(STATUS_CODE_REASON_PHRASE.n200_ok, format!(r#"{{"purged":{n}}}"#).into_bytes()))
    }

    /// A cache kept in files under `dir` (created if missing), bounded to
    /// `max_bytes` in total, oldest dropped first (since 18.11.0). Entries
    /// survive a restart: the directory is indexed when the layer is built,
    /// expired and unreadable files removed. Same rules as
    /// [`memory`](CacheLayer::memory) for what is cached and for how long.
    ///
    /// ```rust,no_run
    /// use rust_web_server::cache::CacheLayer;
    ///
    /// let layer = CacheLayer::disk("/var/cache/rws/api", 512 * 1024 * 1024)?.ttl(600);
    /// # Ok::<(), std::io::Error>(())
    /// ```
    #[cfg(not(target_arch = "wasm32"))]
    pub fn disk(dir: impl Into<std::path::PathBuf>, max_bytes: u64) -> std::io::Result<Self> {
        let store = disk::DiskStore::open(dir, max_bytes)?;
        let mut layer = CacheLayer::memory(0);
        layer.stored.store(store.len(), Ordering::Relaxed);
        layer.disk = Some(Arc::new(store));
        Ok(layer)
    }

    /// The disk store, when this layer has one.
    #[cfg(not(target_arch = "wasm32"))]
    fn disk_store(&self) -> Option<&disk::DiskStore> {
        self.disk.as_deref()
    }

    #[cfg(target_arch = "wasm32")]
    fn disk_store(&self) -> Option<&NoDisk> {
        None
    }

    /// Number of cache hits since the cache was created.
    pub fn hits(&self) -> u64 {
        self.hits.load(Ordering::Relaxed)
    }

    /// Number of cache misses since the cache was created.
    pub fn misses(&self) -> u64 {
        self.misses.load(Ordering::Relaxed)
    }

    /// Current number of entries in the store (including potentially expired ones
    /// that haven't been purged yet).
    pub fn size(&self) -> usize {
        if let Some(disk) = self.disk_store() {
            return disk.len();
        }
        self.store.lock().unwrap().entries.len()
    }

    /// Bytes held by a disk cache's files; `None` for a memory cache.
    pub fn stored_bytes(&self) -> Option<u64> {
        #[cfg(not(target_arch = "wasm32"))]
        return self.disk_store().map(disk::DiskStore::bytes);
        #[cfg(target_arch = "wasm32")]
        None
    }

    /// Remove all entries from the cache.
    pub fn clear(&self) {
        if let Some(disk) = self.disk_store() {
            disk.remove_where(|_| true);
            self.stored.store(0, Ordering::Relaxed);
            return;
        }
        let mut guard = self.store.lock().unwrap();
        guard.entries.clear();
        guard.order.clear();
        self.stored.store(0, Ordering::Relaxed);
    }

    /// Set the time-to-live for cached entries.
    pub fn ttl(mut self, secs: u64) -> Self {
        self.ttl = Duration::from_secs(secs);
        self
    }

    /// Include a request header in the cache key so that different values of
    /// that header produce separate cache entries.
    ///
    /// Header name matching is case-insensitive. Call multiple times to vary
    /// by more than one header.
    ///
    /// ```rust,no_run
    /// use rust_web_server::cache::CacheLayer;
    ///
    /// let layer = CacheLayer::memory(500)
    ///     .vary_by_header("Accept")
    ///     .vary_by_header("Accept-Language");
    /// ```
    pub fn vary_by_header(mut self, name: &str) -> Self {
        self.vary_headers.push(name.to_ascii_lowercase());
        self
    }

    fn store(&self) -> &Mutex<CacheStore> {
        &self.store
    }

    /// Build a cache key from the request URI and any configured vary headers.
    fn cache_key(&self, request: &Request) -> String {
        let mut key = request.request_uri.clone();
        for vh in &self.vary_headers {
            let val = request.headers.iter()
                .find(|h| h.name.eq_ignore_ascii_case(vh))
                .map(|h| h.value.as_ref())
                .unwrap_or("");
            key.push('\x00');
            key.push_str(val);
        }
        key
    }

    /// `true` when the request carries `Cache-Control: no-cache`, meaning the
    /// client wants a fresh response (but we may still store the result).
    fn request_bypasses_cache(request: &Request) -> bool {
        request.headers.iter().any(|h| {
            h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL)
                && h.value.to_ascii_lowercase().contains("no-cache")
        })
    }

    /// `true` for a request whose response may be private to the client
    /// (credentials, cookies) or partial (`Range`): never answered from, nor
    /// stored in, the shared cache (#49).
    fn request_is_private(request: &Request) -> bool {
        request.headers.iter().any(|h| {
            h.name.eq_ignore_ascii_case(Header::_AUTHORIZATION)
                || h.name.eq_ignore_ascii_case(Header::_COOKIE)
                || h.name.eq_ignore_ascii_case(Header::_RANGE)
        })
    }

    /// `true` when the response may be stored and replayed as is:
    /// - a 2xx to a request without `Range` (a 206 is one slice, not the
    ///   resource), held in memory (not streamed), without `Set-Cookie`;
    /// - no `Cache-Control: no-store`, `private` or `no-cache` (`no-cache`
    ///   allows storing only to revalidate before every reuse, which this
    ///   cache can't do — static files are sent with it by default);
    /// - every request header its `Vary` names is part of the cache key, or
    ///   didn't change this response: `Origin` without an
    ///   `Access-Control-Allow-Origin` in it, `Accept-Encoding` with an
    ///   unencoded body (valid for every client). Otherwise e.g. a gzipped
    ///   body would be replayed to a client that didn't ask for gzip.
    ///   `Vary: *` is never cacheable.
    fn response_is_cacheable(&self, request: &Request, response: &Response) -> bool {
        if response.status_code < 200 || response.status_code >= 300 {
            return false;
        }
        if request.header(Header::_RANGE).is_some() {
            return false;
        }
        // A streamed body isn't held in the response: a stored copy would
        // replay an empty body. A cookie-setting response is per client (#49).
        if response.stream_file.is_some() || response.stream_pipe.is_some() {
            return false;
        }
        if response.headers.iter().any(|h| h.name.eq_ignore_ascii_case(Header::_SET_COOKIE)) {
            return false;
        }
        let has = |name: &str| response.headers.iter().any(|h| h.name.eq_ignore_ascii_case(name));
        let varies_harmlessly = |name: &str| {
            (name.eq_ignore_ascii_case(Header::_ORIGIN) && !has(Header::_ACCESS_CONTROL_ALLOW_ORIGIN))
                || (name.eq_ignore_ascii_case(Header::_ACCEPT_ENCODING) && !has(Header::_CONTENT_ENCODING))
        };
        response.headers.iter().all(|h| {
            if h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL) {
                let v = h.value.to_ascii_lowercase();
                !(v.contains("no-store") || v.contains("private") || v.contains("no-cache"))
            } else if h.name.eq_ignore_ascii_case(Header::_VARY) {
                h.value.split(',').map(str::trim).filter(|name| !name.is_empty()).all(|name| {
                    name != "*" && (self.vary_headers.iter().any(|vh| vh.eq_ignore_ascii_case(name)) || varies_harmlessly(name))
                })
            } else {
                true
            }
        })
    }

    /// Age of the entry in whole seconds, capped at u64::MAX.
    fn age_secs(entry: &CachedEntry) -> u64 {
        entry.inserted_at.elapsed().as_secs()
    }

    /// Build a response from a cache hit, injecting an `Age` header.
    fn cached_response(entry: &CachedEntry) -> Response {
        let mut response = entry.response.clone();
        let age = Self::age_secs(entry);
        // Replace or add the Age header.
        if let Some(h) = response.headers.iter_mut().find(|h| h.name.eq_ignore_ascii_case("Age")) {
            h.value = age.to_string().into();
        } else {
            response.headers.push(Header { name: "Age".into(), value: age.to_string().into() });
        }
        response
    }
}

impl Middleware for CacheLayer {
    /// Headers only: a streamed body passes through untouched.
    fn accepts_streamed_body(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
        true
    }


    fn handle(
        &self,
        request: &Request,
        connection: &ConnectionInfo,
        next: &dyn Application,
    ) -> Result<Response, String> {
        if let Some(answer) = self.answer_purge(request) {
            return Ok(answer);
        }
        // Only cache GET requests, and never private or partial ones.
        if request.method != METHOD.get || Self::request_is_private(request) {
            return next.execute(request, connection);
        }

        let bypass = Self::request_bypasses_cache(request);

        if let Some(disk) = self.disk_store() {
            if !bypass && !disk.is_empty() {
                let key = self.cache_key(request);
                if let Some((mut response, age)) = disk.get(&key) {
                    self.hits.fetch_add(1, Ordering::Relaxed);
                    response.headers.retain(|h| !h.name.eq_ignore_ascii_case("Age"));
                    response.headers.push(Header { name: "Age".into(), value: age.as_secs().to_string().into() });
                    return Ok(response);
                }
            }
            self.misses.fetch_add(1, Ordering::Relaxed);
            let response = next.execute(request, connection)?;
            let ttl = freshness(&response, self.ttl);
            if !ttl.is_zero() && self.response_is_cacheable(request, &response) {
                disk.put(&self.cache_key(request), &response, ttl);
                self.stored.store(disk.len(), Ordering::Relaxed);
            }
            return Ok(response);
        }

        if !bypass && self.stored.load(Ordering::Relaxed) > 0 {
            // Check for a valid cache hit.
            let key = self.cache_key(request);
            let guard = self.store().lock().unwrap();
            if let Some(entry) = guard.get(&key) {
                self.hits.fetch_add(1, Ordering::Relaxed);
                return Ok(Self::cached_response(entry));
            }
        }
        self.misses.fetch_add(1, Ordering::Relaxed);

        // Cache miss (or bypass): call the handler.
        let response = next.execute(request, connection)?;

        let ttl = freshness(&response, self.ttl);
        if !ttl.is_zero() && self.response_is_cacheable(request, &response) {
            let key = self.cache_key(request);
            let mut guard = self.store().lock().unwrap();
            guard.purge_expired();
            guard.insert(
                key,
                CachedEntry { response: response.clone(), inserted_at: Instant::now(), ttl },
                self.capacity,
            );
            self.stored.store(guard.entries.len(), Ordering::Relaxed);
        }

        Ok(response)
    }

    /// In-memory: safe on the event loop. A disk store reads and writes
    /// files, so it runs on the blocking pool.
    fn runs_inline(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
        self.disk_store().is_none()
    }
}

/// Stand-in for the disk store where there is none (wasm32).
#[cfg(target_arch = "wasm32")]
struct NoDisk;

#[cfg(target_arch = "wasm32")]
impl NoDisk {
    fn len(&self) -> usize { 0 }
    fn is_empty(&self) -> bool { true }
    fn remove_where(&self, _matches: impl Fn(&str) -> bool) -> usize { 0 }
    fn get(&self, _key: &str) -> Option<(Response, Duration)> { None }
    fn put(&self, _key: &str, _response: &Response, _ttl: Duration) {}
}