tower_http_cache/backend/
mod.rs

1//! Storage backends for the cache layer.
2//!
3//! The cache layer requires a [`CacheBackend`] implementation to persist
4//! cached responses. This module ships with:
5//! - [`memory::InMemoryBackend`] — a fast, process-local cache backed by [`moka`].
6//! - `redis::RedisBackend` *(optional)* — a distributed cache when the
7//!   `redis-backend` crate feature is enabled.
8//! - `memcached::MemcachedBackend` *(optional)* — a distributed cache when the
9//!   `memcached-backend` crate feature is enabled.
10//!
11//! Backends are responsible for answering cache lookups, storing entries,
12//! and enforcing per-entry stale windows.
13
14#[cfg(feature = "memcached-backend")]
15pub mod memcached;
16#[cfg(feature = "in-memory")]
17pub mod memory;
18pub mod multi_tier;
19#[cfg(feature = "redis-backend")]
20pub mod redis;
21
22use async_trait::async_trait;
23use bytes::Bytes;
24use http::{HeaderName, HeaderValue, Response, StatusCode, Version};
25use std::time::{Duration, SystemTime};
26
27use crate::error::CacheError;
28use crate::layer::SyncBoxBody;
29
30/// Cached response payload captured by the cache layer.
31#[derive(Debug, Clone)]
32#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
33pub struct CacheEntry {
34    #[cfg_attr(feature = "serde", serde(with = "status_code_serde"))]
35    pub status: StatusCode,
36    #[cfg_attr(feature = "serde", serde(with = "version_serde"))]
37    pub version: Version,
38    pub headers: Vec<(String, Vec<u8>)>,
39    #[cfg_attr(feature = "serde", serde(with = "bytes_serde"))]
40    pub body: Bytes,
41    pub tags: Option<Vec<String>>,
42}
43
44// Custom serde helpers for http types
45#[cfg(feature = "serde")]
46mod status_code_serde {
47    use http::StatusCode;
48    use serde::{Deserialize, Deserializer, Serialize, Serializer};
49
50    pub fn serialize<S>(status: &StatusCode, serializer: S) -> Result<S::Ok, S::Error>
51    where
52        S: Serializer,
53    {
54        status.as_u16().serialize(serializer)
55    }
56
57    pub fn deserialize<'de, D>(deserializer: D) -> Result<StatusCode, D::Error>
58    where
59        D: Deserializer<'de>,
60    {
61        let code = u16::deserialize(deserializer)?;
62        StatusCode::from_u16(code).map_err(serde::de::Error::custom)
63    }
64}
65
66#[cfg(feature = "serde")]
67mod version_serde {
68    use http::Version;
69    use serde::{Deserialize, Deserializer, Serialize, Serializer};
70
71    pub fn serialize<S>(version: &Version, serializer: S) -> Result<S::Ok, S::Error>
72    where
73        S: Serializer,
74    {
75        let v = match *version {
76            Version::HTTP_09 => 0,
77            Version::HTTP_10 => 1,
78            Version::HTTP_11 => 2,
79            Version::HTTP_2 => 3,
80            Version::HTTP_3 => 4,
81            _ => 5,
82        };
83        v.serialize(serializer)
84    }
85
86    pub fn deserialize<'de, D>(deserializer: D) -> Result<Version, D::Error>
87    where
88        D: Deserializer<'de>,
89    {
90        let v = u8::deserialize(deserializer)?;
91        Ok(match v {
92            0 => Version::HTTP_09,
93            1 => Version::HTTP_10,
94            2 => Version::HTTP_11,
95            3 => Version::HTTP_2,
96            4 => Version::HTTP_3,
97            _ => Version::HTTP_11, // Default fallback
98        })
99    }
100}
101
102#[cfg(feature = "serde")]
103mod bytes_serde {
104    use bytes::Bytes;
105    use serde::{Deserialize, Deserializer, Serializer};
106
107    pub fn serialize<S>(bytes: &Bytes, serializer: S) -> Result<S::Ok, S::Error>
108    where
109        S: Serializer,
110    {
111        serializer.serialize_bytes(bytes)
112    }
113
114    pub fn deserialize<'de, D>(deserializer: D) -> Result<Bytes, D::Error>
115    where
116        D: Deserializer<'de>,
117    {
118        let vec = Vec::<u8>::deserialize(deserializer)?;
119        Ok(Bytes::from(vec))
120    }
121}
122
123impl CacheEntry {
124    /// Creates a new cached response entry.
125    ///
126    /// The entry captures the response status, HTTP version, a serialized
127    /// subset of headers, and the collected response body.
128    pub fn new(
129        status: StatusCode,
130        version: Version,
131        headers: Vec<(String, Vec<u8>)>,
132        body: Bytes,
133    ) -> Self {
134        Self {
135            status,
136            version,
137            headers,
138            body,
139            tags: None,
140        }
141    }
142
143    /// Creates a new cached response entry with tags.
144    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
145        self.tags = Some(tags);
146        self
147    }
148
149    /// Converts the entry back into an `http::Response`.
150    pub fn into_response(self) -> Response<SyncBoxBody> {
151        use http_body_util::BodyExt;
152
153        let full_body = http_body_util::Full::from(self.body);
154        let boxed_body = full_body
155            .map_err(|never| -> Box<dyn std::error::Error + Send + Sync> { match never {} })
156            .boxed();
157
158        let mut response = Response::new(SyncBoxBody::new(boxed_body));
159        *response.status_mut() = self.status;
160        *response.version_mut() = self.version;
161
162        let headers = response.headers_mut();
163        headers.clear();
164        for (name, value) in self.headers {
165            if let (Ok(name), Ok(value)) = (
166                HeaderName::from_bytes(name.as_bytes()),
167                HeaderValue::from_bytes(&value),
168            ) {
169                headers.append(name, value);
170            }
171        }
172
173        response
174    }
175}
176
177#[derive(Debug, Clone)]
178pub struct CacheRead {
179    /// Cached entry together with timing metadata.
180    pub entry: CacheEntry,
181    pub expires_at: Option<SystemTime>,
182    pub stale_until: Option<SystemTime>,
183}
184
185#[async_trait]
186pub trait CacheBackend: Send + Sync + Clone + 'static {
187    /// Fetches a cached entry by key.
188    ///
189    /// Returns `Ok(None)` when the backend does not have a value or the
190    /// entry has expired.
191    async fn get(&self, key: &str) -> Result<Option<CacheRead>, CacheError>;
192
193    /// Stores an entry with a time-to-live and additional stale window.
194    async fn set(
195        &self,
196        key: String,
197        entry: CacheEntry,
198        ttl: Duration,
199        stale_for: Duration,
200    ) -> Result<(), CacheError>;
201
202    /// Invalidates the cache entry for `key`, if present.
203    async fn invalidate(&self, key: &str) -> Result<(), CacheError>;
204
205    /// Retrieves all cache keys associated with a tag.
206    ///
207    /// Returns an empty vector if tags are not supported by this backend.
208    async fn get_keys_by_tag(&self, _tag: &str) -> Result<Vec<String>, CacheError> {
209        Ok(Vec::new())
210    }
211
212    /// Invalidates all cache entries associated with a tag.
213    ///
214    /// Returns the number of entries invalidated.
215    async fn invalidate_by_tag(&self, tag: &str) -> Result<usize, CacheError> {
216        let keys = self.get_keys_by_tag(tag).await?;
217        let count = keys.len();
218        for key in keys {
219            let _ = self.invalidate(&key).await;
220        }
221        Ok(count)
222    }
223
224    /// Invalidates all cache entries associated with multiple tags.
225    ///
226    /// Returns the total number of entries invalidated (may include duplicates).
227    async fn invalidate_by_tags(&self, tags: &[String]) -> Result<usize, CacheError> {
228        let mut total = 0;
229        for tag in tags {
230            total += self.invalidate_by_tag(tag).await?;
231        }
232        Ok(total)
233    }
234
235    /// Lists all currently indexed tags.
236    ///
237    /// Returns an empty vector if tags are not supported by this backend.
238    async fn list_tags(&self) -> Result<Vec<String>, CacheError> {
239        Ok(Vec::new())
240    }
241}
tower_http_cache/backend/mod.rs

tower_http_cache/backend/
mod.rs
