featherbit/stores/
redis_cache.rs

1//! Shared response cache over a declared `stores:` entry.
2//!
3//! One redis hash per entry: `status`, `headers` (JSON) and `body` (raw
4//! bytes). Redis values are binary-safe, so the body is stored as-is --
5//! base64 would inflate every cached response by a third, and a JSON wrapper
6//! would escape it on top of that.
7
8use std::collections::HashMap;
9use std::sync::Arc;
10use std::time::Duration;
11
12use async_trait::async_trait;
13use bytes::Bytes;
14use redis::AsyncCommands;
15
16use crate::stores::namespaces;
17use crate::stores::redis_store::RedisStoreClient;
18use crate::traffic::{CacheError, CachedResponse, ResponseCache};
19
20/// The redis key for a cache entry.
21pub(crate) fn cache_key(prefix: &str, key: &str) -> String {
22    format!("{}:{}:{}", prefix, namespaces::CACHE, key)
23}
24
25/// Backslash-escapes every glob metacharacter Redis's `MATCH` understands.
26///
27/// An `id` is free-form config text. Without this, a pair named `a*` would
28/// purge every pair beginning with `a`.
29pub(crate) fn escape_glob(s: &str) -> String {
30    let mut out = String::with_capacity(s.len());
31    for c in s.chars() {
32        if matches!(c, '*' | '?' | '[' | ']' | '\\') {
33            out.push('\\');
34        }
35        out.push(c);
36    }
37    out
38}
39
40/// Keys per `UNLINK`. Small enough that no single round trip holds the
41/// server long; large enough that a big purge is not thousands of them.
42const UNLINK_BATCH: usize = 200;
43
44pub struct RedisResponseCache {
45    pub(crate) client: Arc<RedisStoreClient>,
46}
47
48impl RedisResponseCache {
49    pub fn new(client: Arc<RedisStoreClient>) -> Self {
50        Self { client }
51    }
52
53    pub(crate) fn redis_key(&self, key: &str) -> String {
54        cache_key(self.client.key_prefix(), key)
55    }
56}
57
58#[async_trait]
59impl ResponseCache for RedisResponseCache {
60    async fn get(&self, key: &str) -> Result<Option<CachedResponse>, CacheError> {
61        let mut conn = self.client.conn().await.map_err(CacheError)?;
62        let fields: HashMap<String, Vec<u8>> = conn
63            .hgetall(self.redis_key(key))
64            .await
65            .map_err(|e| CacheError(e.to_string()))?;
66
67        if fields.is_empty() {
68            return Ok(None);
69        }
70
71        // A hash the gateway cannot parse is unusable either way, so it reads
72        // as a miss rather than an error: the request keeps moving, and the
73        // two outcomes stay distinguishable in metrics.
74        let (Some(status), Some(headers), Some(body)) = (
75            fields.get("status"),
76            fields.get("headers"),
77            fields.get("body"),
78        ) else {
79            return Ok(None);
80        };
81        let Ok(status) = String::from_utf8_lossy(status).parse::<u16>() else {
82            return Ok(None);
83        };
84        let Ok(headers) = serde_json::from_slice::<HashMap<String, Vec<String>>>(headers) else {
85            return Ok(None);
86        };
87
88        Ok(Some(CachedResponse {
89            status,
90            headers,
91            body: Bytes::copy_from_slice(body),
92        }))
93    }
94
95    async fn put(
96        &self,
97        key: &str,
98        entry: &CachedResponse,
99        ttl: Duration,
100    ) -> Result<(), CacheError> {
101        let mut conn = self.client.conn().await.map_err(CacheError)?;
102        let redis_key = self.redis_key(key);
103        let headers = serde_json::to_vec(&entry.headers).map_err(|e| CacheError(e.to_string()))?;
104
105        // Fields and expiry in one pipeline: an entry without a TTL would
106        // outlive its freshness and never be swept.
107        redis::pipe()
108            .atomic()
109            .hset(&redis_key, "status", entry.status.to_string())
110            .hset(&redis_key, "headers", headers)
111            .hset(&redis_key, "body", entry.body.as_ref())
112            .expire(&redis_key, ttl.as_secs().max(1) as i64)
113            .query_async::<()>(&mut conn)
114            .await
115            .map_err(|e| CacheError(e.to_string()))
116    }
117
118    async fn purge(&self, id: &str) -> Result<u64, CacheError> {
119        let mut conn = self.client.conn().await.map_err(CacheError)?;
120        // Escape the whole computed literal once, not just `id`: the
121        // store's `key_prefix` is also free-form config text, and a glob
122        // metacharacter in it must not widen the SCAN MATCH either.
123        // `\u{1}` and `:` are not glob metacharacters, so this leaves the
124        // pattern for a clean prefix unchanged.
125        let literal = self.redis_key(&crate::traffic::cache::pair_prefix(id));
126        let pattern = format!("{}*", escape_glob(&literal));
127
128        // SCAN driven by hand rather than through `redis::AsyncIter`: that
129        // type is deprecated without the `safe_iterators` feature and fails
130        // `-D warnings`. The explicit loop also makes the batching visible.
131        let mut cursor: u64 = 0;
132        let mut removed: u64 = 0;
133        loop {
134            let (next, keys): (u64, Vec<String>) = redis::cmd("SCAN")
135                .arg(cursor)
136                .arg("MATCH")
137                .arg(&pattern)
138                .arg("COUNT")
139                .arg(UNLINK_BATCH)
140                .query_async(&mut conn)
141                .await
142                .map_err(|e| CacheError(e.to_string()))?;
143
144            for chunk in keys.chunks(UNLINK_BATCH) {
145                let n: u64 = redis::cmd("UNLINK")
146                    .arg(chunk)
147                    .query_async(&mut conn)
148                    .await
149                    .map_err(|e| CacheError(e.to_string()))?;
150                removed += n;
151            }
152
153            if next == 0 {
154                break;
155            }
156            cursor = next;
157        }
158        Ok(removed)
159    }
160}
161
162#[cfg(all(test, feature = "redis-store"))]
163mod tests {
164    use super::*;
165
166    /// An `id` is free-form config text. Glob metacharacters in it must not
167    /// widen a SCAN MATCH: purging the pair literally named `a*` must not
168    /// match every pair beginning with `a`.
169    #[test]
170    fn test_escape_glob_neutralises_every_metacharacter() {
171        assert_eq!(escape_glob("plain"), "plain");
172        assert_eq!(escape_glob("a*b?c[d]e\\f"), "a\\*b\\?c\\[d\\]e\\\\f");
173    }
174
175    fn store_url() -> Option<String> {
176        std::env::var("FEATHERBIT_TEST_REDIS_URL")
177            .ok()
178            .filter(|s| !s.is_empty())
179    }
180
181    fn client(url: &str) -> Arc<crate::stores::redis_store::RedisStoreClient> {
182        let cfg: crate::config::StoreConfig = serde_yaml::from_str(&format!(
183            "name: cache-test\ntype: redis\nurl: {url}\nkey_prefix: fbtest\n"
184        ))
185        .unwrap();
186        Arc::new(crate::stores::redis_store::RedisStoreClient::build(&cfg).unwrap())
187    }
188
189    fn response(body: &[u8]) -> CachedResponse {
190        let mut headers = HashMap::new();
191        headers.insert(
192            "x-multi".to_string(),
193            vec!["a".to_string(), "b".to_string()],
194        );
195        CachedResponse {
196            status: 203,
197            headers,
198            body: Bytes::copy_from_slice(body),
199        }
200    }
201
202    /// The point of the whole feature: an entry written by one instance is
203    /// readable by another. Two handles over the same store stand in for two
204    /// gateways -- nothing else demonstrates sharing.
205    #[tokio::test]
206    async fn test_an_entry_written_by_one_handle_is_read_by_another() {
207        let Some(url) = store_url() else { return };
208        let writer = RedisResponseCache::new(client(&url));
209        let reader = RedisResponseCache::new(client(&url));
210        let key = format!("share-{}", uuid::Uuid::new_v4());
211
212        writer
213            .put(&key, &response(b"shared"), Duration::from_secs(60))
214            .await
215            .unwrap();
216
217        let got = reader
218            .get(&key)
219            .await
220            .unwrap()
221            .expect("the second handle must see it");
222        assert_eq!(got.body, Bytes::from_static(b"shared"));
223    }
224
225    /// Binary bodies and multi-value headers must survive the encoding: a
226    /// base64 or JSON wrapper would be the easy way to get this wrong.
227    #[tokio::test]
228    async fn test_a_binary_body_and_multi_value_header_round_trip() {
229        let Some(url) = store_url() else { return };
230        let cache = RedisResponseCache::new(client(&url));
231        let key = format!("bin-{}", uuid::Uuid::new_v4());
232        let body = vec![0u8, 159, 146, 150, 255];
233
234        cache
235            .put(&key, &response(&body), Duration::from_secs(60))
236            .await
237            .unwrap();
238
239        let got = cache.get(&key).await.unwrap().unwrap();
240        assert_eq!(got.status, 203);
241        assert_eq!(got.body.as_ref(), body.as_slice());
242        assert_eq!(
243            got.headers.get("x-multi").unwrap(),
244            &vec!["a".to_string(), "b".to_string()]
245        );
246    }
247
248    #[tokio::test]
249    async fn test_a_missing_key_is_ok_none() {
250        let Some(url) = store_url() else { return };
251        let cache = RedisResponseCache::new(client(&url));
252        assert!(cache
253            .get(&format!("absent-{}", uuid::Uuid::new_v4()))
254            .await
255            .unwrap()
256            .is_none());
257    }
258
259    /// Expiry is redis's own TTL, never an Instant comparison -- two instances
260    /// do not share a monotonic clock.
261    #[tokio::test]
262    async fn test_the_key_carries_a_ttl() {
263        let Some(url) = store_url() else { return };
264        let cache = RedisResponseCache::new(client(&url));
265        let key = format!("ttl-{}", uuid::Uuid::new_v4());
266        cache
267            .put(&key, &response(b"x"), Duration::from_secs(30))
268            .await
269            .unwrap();
270
271        let mut conn = cache.client.conn().await.unwrap();
272        let ttl: i64 = redis::cmd("TTL")
273            .arg(cache.redis_key(&key))
274            .query_async(&mut conn)
275            .await
276            .unwrap();
277        assert!(
278            ttl > 0,
279            "the entry must expire on redis's clock, not ours: {ttl}"
280        );
281    }
282
283    /// More entries than one SCAN page, so the cursor loop and the UNLINK
284    /// batching are actually exercised, and a sibling pair to prove the
285    /// boundary holds on the shared backend too.
286    #[tokio::test]
287    async fn test_redis_purge_removes_only_the_named_pair_across_scan_pages() {
288        let Some(url) = store_url() else { return };
289        let cache = RedisResponseCache::new(client(&url));
290        let run = uuid::Uuid::new_v4();
291        let target = format!("purge-{run}");
292        let sibling = format!("purge-{run}-v2");
293        let ttl = Duration::from_secs(60);
294
295        for i in 0..250 {
296            cache
297                .put(&format!("{target}\u{1}/{i}"), &response(b"x"), ttl)
298                .await
299                .unwrap();
300        }
301        cache
302            .put(&format!("{sibling}\u{1}/0"), &response(b"keep"), ttl)
303            .await
304            .unwrap();
305
306        let removed = cache.purge(&target).await.unwrap();
307
308        assert_eq!(
309            removed, 250,
310            "every entry of the pair, across more than one SCAN page"
311        );
312        assert!(cache
313            .get(&format!("{target}\u{1}/0"))
314            .await
315            .unwrap()
316            .is_none());
317        assert!(
318            cache
319                .get(&format!("{sibling}\u{1}/0"))
320                .await
321                .unwrap()
322                .is_some(),
323            "the sibling pair must be untouched"
324        );
325    }
326}
featherbit/stores/redis_cache.rs

featherbit/stores/
redis_cache.rs
