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

rustlavel_cache/
lib.rs
