rustlavel_cache/
lib.rs

1//! rustlavel-cache: caching and rate limiting.
2//!
3//! One [`Cache`] trait, three drivers, and a rate limiter built on top of it:
4//!
5//! | driver   | shared between processes | survives a restart | needs |
6//! |----------|--------------------------|--------------------|-------|
7//! | `memory` | no                       | no                 | nothing |
8//! | `file`   | between processes on one host | yes           | a directory |
9//! | `redis`  | yes                      | yes                | a Redis server |
10//!
11//! ```ignore
12//! use rustlavel_cache::{Cache, CacheExt, CacheStore, Throttle};
13//!
14//! let cache = CacheStore::from_config(&config)?;
15//!
16//! let users = cache
17//!     .remember("users:active", Duration::from_secs(300), || async {
18//!         Ok(load_active_users().await?)
19//!     })
20//!     .await?;
21//!
22//! router.middleware(Throttle::per_minute(&cache, 60));
23//! ```
24//!
25//! The Redis client is written from scratch — RESP encoder and decoder,
26//! connection, handshake and pool, all on Tokio's TCP — for the same reason the
27//! HTTP server and the PostgreSQL driver are: a framework that owns its wire
28//! protocols owns its error messages, its performance and its security
29//! posture. See [`redis::resp`] for the protocol itself.
30//!
31//! Every lookup dispatches `cache.hit` or `cache.miss` on
32//! [`rustlavel_core::events`], so Telescope can show a hit rate without this
33//! crate knowing Telescope exists. Nothing is built when no subscriber is
34//! listening.
35
36pub mod config;
37pub mod file;
38pub mod idempotency;
39pub mod memory;
40pub mod rate_limit;
41pub mod redis;
42pub mod store;
43pub mod throttle;
44
45pub use config::{CacheConfig, CacheStore, Driver};
46pub use file::FileStore;
47pub use idempotency::Idempotency;
48pub use memory::MemoryStore;
49pub use rate_limit::{RateLimit, RateLimiter};
50pub use redis::{RedisConfig, RedisStore};
51pub use store::{BoxFuture, Cache, CacheExt};
52pub use throttle::Throttle;
53
54pub use rustlavel_core::{Error, Json, Result};
55
56/// What an application importing this crate usually wants.
57pub mod prelude {
58    pub use crate::{Cache, CacheExt, CacheStore, Idempotency, RateLimiter, Throttle};
59    pub use rustlavel_core::{Json, Result};
60}
61
62#[cfg(test)]
63mod tests {
64    //! The behavioural suite every driver must satisfy.
65    //!
66    //! It is written once against `dyn Cache` and run against each driver, so a
67    //! driver cannot quietly disagree with the others about what `forget`
68    //! returns or what an expired key looks like. `tests/redis.rs` holds the
69    //! Redis driver to the same contract against a live server; it carries its
70    //! own copy because an integration test links the crate without `cfg(test)`.
71
72    use super::*;
73    use std::sync::Arc;
74    use std::time::Duration;
75
76    /// Assert the full contract against one driver.
77    async fn assert_cache_contract(cache: &dyn Cache) {
78        cache.flush().await.unwrap();
79
80        // A miss is None, not an error.
81        assert_eq!(cache.get("absent").await.unwrap(), None);
82        assert!(!cache.has("absent").await.unwrap());
83        assert!(!cache.forget("absent").await.unwrap());
84
85        // put / get round-trips every JSON shape.
86        for value in [
87            Json::Null,
88            Json::from(true),
89            Json::from(-17),
90            Json::from(1.5),
91            Json::from("a string with \" and \\ and \n in it"),
92            Json::from(vec![1, 2, 3]),
93            Json::object([("nested", Json::object([("deep", Json::from(true))]))]),
94        ] {
95            cache.put("shape", value.clone(), Duration::from_secs(60)).await.unwrap();
96            assert_eq!(cache.get("shape").await.unwrap(), Some(value.clone()), "round trip failed");
97        }
98
99        // forever survives without a TTL.
100        cache.forever("immortal", Json::from("forever")).await.unwrap();
101        assert_eq!(cache.ttl("immortal").await.unwrap(), None);
102        assert!(cache.has("immortal").await.unwrap());
103
104        // forget reports whether it removed something.
105        assert!(cache.forget("immortal").await.unwrap());
106        assert!(!cache.forget("immortal").await.unwrap());
107
108        // TTL actually expires.
109        cache.put("brief", Json::from("gone soon"), Duration::from_millis(120)).await.unwrap();
110        assert!(cache.has("brief").await.unwrap());
111        assert!(cache.ttl("brief").await.unwrap().is_some());
112        tokio::time::sleep(Duration::from_millis(220)).await;
113        assert_eq!(cache.get("brief").await.unwrap(), None, "the TTL did not expire the key");
114        assert!(!cache.has("brief").await.unwrap());
115
116        // increment / decrement.
117        assert_eq!(cache.increment("counter", 1).await.unwrap(), 1);
118        assert_eq!(cache.increment("counter", 4).await.unwrap(), 5);
119        assert_eq!(cache.decrement("counter", 2).await.unwrap(), 3);
120        assert_eq!(cache.get("counter").await.unwrap(), Some(Json::from(3)));
121        assert_eq!(cache.decrement("fresh-counter", 3).await.unwrap(), -3);
122
123        // increment_within starts the window only once.
124        assert_eq!(cache.increment_within("window", 1, Duration::from_secs(60)).await.unwrap(), 1);
125        assert_eq!(cache.increment_within("window", 1, Duration::from_secs(60)).await.unwrap(), 2);
126        let remaining = cache.ttl("window").await.unwrap().expect("a window has a deadline");
127        assert!(remaining <= Duration::from_secs(60));
128
129        // remember computes on a miss and only on a miss.
130        let computed = cache
131            .remember("remembered", Duration::from_secs(60), || async { Ok(Json::from("first")) })
132            .await
133            .unwrap();
134        assert_eq!(computed, Json::from("first"));
135
136        let cached = cache
137            .remember("remembered", Duration::from_secs(60), || async {
138                panic!("remember must not recompute a hit")
139            })
140            .await
141            .unwrap();
142        assert_eq!(cached, Json::from("first"));
143
144        // remember_forever likewise.
145        cache
146            .remember_forever("remembered-forever", || async { Ok(Json::from(7)) })
147            .await
148            .unwrap();
149        assert_eq!(cache.ttl("remembered-forever").await.unwrap(), None);
150
151        // pull returns the value and leaves nothing behind.
152        assert_eq!(cache.pull("remembered").await.unwrap(), Some(Json::from("first")));
153        assert_eq!(cache.pull("remembered").await.unwrap(), None);
154
155        // A zero TTL means "already expired".
156        cache.forever("doomed", Json::from(1)).await.unwrap();
157        cache.put("doomed", Json::from(2), Duration::ZERO).await.unwrap();
158        assert!(!cache.has("doomed").await.unwrap());
159
160        // flush empties everything.
161        cache.forever("a", Json::from(1)).await.unwrap();
162        cache.forever("b", Json::from(2)).await.unwrap();
163        cache.flush().await.unwrap();
164        assert_eq!(cache.get("a").await.unwrap(), None);
165        assert_eq!(cache.get("b").await.unwrap(), None);
166        assert_eq!(cache.get("counter").await.unwrap(), None);
167    }
168
169    #[tokio::test]
170    async fn the_memory_driver_satisfies_the_cache_contract() {
171        assert_cache_contract(&MemoryStore::new()).await;
172    }
173
174    #[tokio::test]
175    async fn the_file_driver_satisfies_the_cache_contract() {
176        // Its own directory: the contract calls `flush`, which would wipe a
177        // concurrently running test sharing the same one.
178        let directory = std::env::temp_dir()
179            .join(format!("rustlavel-cache-contract-{}", std::process::id()));
180        let _ = std::fs::remove_dir_all(&directory);
181
182        assert_cache_contract(&FileStore::new(&directory).unwrap()).await;
183
184        let _ = std::fs::remove_dir_all(&directory);
185    }
186
187    #[tokio::test]
188    async fn a_boxed_driver_satisfies_the_contract_too() {
189        // Proves the trait really is dyn-compatible end to end, which is what
190        // the whole `BoxFuture` return style buys.
191        let cache: Arc<dyn Cache> = Arc::new(MemoryStore::new());
192        assert_cache_contract(cache.as_ref()).await;
193        assert_eq!(cache.driver(), "memory");
194    }
195
196    #[tokio::test]
197    async fn a_lookup_dispatches_a_hit_or_a_miss_event() {
198        use rustlavel_core::events::{self, Event};
199        use std::sync::Mutex;
200
201        // The event registry is process-global and tests run concurrently, so
202        // this one listens for keys only it uses rather than assuming it is
203        // the only thing touching a cache right now.
204        let marker = "event-probe:";
205        events::clear_subscribers();
206
207        let seen: Arc<Mutex<Vec<(String, String)>>> = Arc::new(Mutex::new(Vec::new()));
208        let sink = Arc::clone(&seen);
209        events::subscribe(move |event: &Event| {
210            let key = event.field("key").and_then(Json::as_str).unwrap_or_default();
211            if event.kind.starts_with("cache.") && key.starts_with(marker) {
212                sink.lock().unwrap().push((event.kind.to_string(), key.to_string()));
213            }
214        });
215
216        let cache = MemoryStore::new();
217        cache.get("event-probe:missing").await.unwrap();
218        cache.forever("event-probe:present", Json::from(1)).await.unwrap();
219        cache.get("event-probe:present").await.unwrap();
220
221        // A driver must report its own name, so Telescope can tell stores apart.
222        let names: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
223        let slot = Arc::clone(&names);
224        events::subscribe(move |event: &Event| {
225            if event.kind == "cache.miss" {
226                *slot.lock().unwrap() =
227                    event.field("store").and_then(Json::as_str).map(str::to_string);
228            }
229        });
230        cache.get("event-probe:another-miss").await.unwrap();
231        let store_name = names.lock().unwrap().clone();
232
233        let recorded = seen.lock().unwrap().clone();
234        events::clear_subscribers();
235
236        assert_eq!(store_name.as_deref(), Some("memory"));
237        assert_eq!(
238            recorded,
239            vec![
240                ("cache.miss".to_string(), "event-probe:missing".to_string()),
241                ("cache.hit".to_string(), "event-probe:present".to_string()),
242                ("cache.miss".to_string(), "event-probe:another-miss".to_string()),
243            ]
244        );
245    }
246}
rustlavel_cache/lib.rs

rustlavel_cache/
lib.rs
