camel_core/cache/
memory.rs

1//! In-memory cache backend backed by [`moka`].
2//!
3//! Size-eviction only. No TTL-based expiration is configured at the
4//! moka level — expiry is checked in-band during [`CacheRepository::get`].
5
6use std::sync::Arc;
7use std::sync::atomic::AtomicU64;
8use std::sync::atomic::Ordering;
9use std::time::Duration;
10use std::time::SystemTime;
11
12use async_trait::async_trait;
13use camel_api::CamelError;
14use camel_api::cache::CacheEntry;
15use camel_api::cache::CacheRepository;
16use camel_api::cache::CacheStats;
17
18/// In-memory cache repository with size-based eviction.
19///
20/// Uses [`moka::future::Cache`] with a max-capacity bound. Expiry is
21/// checked in-band in [`get`](CacheRepository::get) — no
22/// `expire_after` or `time_to_live` is configured on the moka cache.
23pub struct MemoryCacheRepository {
24    name: String,
25    inner: moka::future::Cache<String, CacheEntry>,
26    hits: Arc<AtomicU64>,
27    misses: Arc<AtomicU64>,
28    evictions: Arc<AtomicU64>,
29    peek_stale_served: Arc<AtomicU64>,
30    invalidations: Arc<AtomicU64>,
31}
32
33impl std::fmt::Debug for MemoryCacheRepository {
34    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        f.debug_struct("MemoryCacheRepository")
36            .field("name", &self.name)
37            .field("stats", &self.stats_snapshot())
38            .finish()
39    }
40}
41
42impl MemoryCacheRepository {
43    /// Create a new memory-backed cache repository.
44    ///
45    /// `max_capacity` is the maximum number of entries before eviction.
46    pub fn new(name: impl Into<String>, max_capacity: usize) -> Self {
47        let hits = Arc::new(AtomicU64::new(0));
48        let misses = Arc::new(AtomicU64::new(0));
49        let evictions = Arc::new(AtomicU64::new(0));
50        let peek_stale_served = Arc::new(AtomicU64::new(0));
51        let invalidations = Arc::new(AtomicU64::new(0));
52        let evictions_clone = Arc::clone(&evictions);
53
54        let inner = moka::future::CacheBuilder::new(max_capacity as u64)
55            .eviction_listener(move |_key, _value, cause| {
56                // Only count capacity-driven size evictions — NOT explicit invalidate() calls.
57                if cause == moka::notification::RemovalCause::Size {
58                    evictions_clone.fetch_add(1, Ordering::Relaxed);
59                }
60            })
61            .build();
62
63        Self {
64            name: name.into(),
65            inner,
66            hits,
67            misses,
68            evictions,
69            peek_stale_served,
70            invalidations,
71        }
72    }
73
74    /// Synchronous snapshot of the current stats, shared by the async trait
75    /// method and the [`std::fmt::Debug`] impl (which cannot await).
76    fn stats_snapshot(&self) -> CacheStats {
77        CacheStats {
78            hits: self.hits.load(Ordering::Relaxed),
79            misses: self.misses.load(Ordering::Relaxed),
80            evictions: self.evictions.load(Ordering::Relaxed),
81            entries: self.inner.entry_count(),
82            peek_stale_served: self.peek_stale_served.load(Ordering::Relaxed),
83            invalidations: self.invalidations.load(Ordering::Relaxed),
84            bytes: None,
85        }
86    }
87}
88
89#[async_trait]
90impl CacheRepository for MemoryCacheRepository {
91    fn name(&self) -> &str {
92        &self.name
93    }
94
95    async fn get(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
96        match self.inner.get(key).await {
97            Some(entry) => {
98                let expired = entry
99                    .expires_at
100                    .map(|e| e <= SystemTime::now())
101                    .unwrap_or(false);
102                if expired {
103                    self.misses.fetch_add(1, Ordering::Relaxed);
104                    Ok(None)
105                } else {
106                    self.hits.fetch_add(1, Ordering::Relaxed);
107                    Ok(Some(entry))
108                }
109            }
110            None => {
111                self.misses.fetch_add(1, Ordering::Relaxed);
112                Ok(None)
113            }
114        }
115    }
116
117    async fn set(
118        &self,
119        key: &str,
120        mut value: CacheEntry,
121        ttl: Option<Duration>,
122    ) -> Result<(), CamelError> {
123        value.expires_at = ttl.map(|d| SystemTime::now() + d);
124        self.inner.insert(key.to_string(), value).await;
125        Ok(())
126    }
127
128    async fn peek_stale(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
129        let entry = self.inner.get(key).await;
130        if entry.is_some() {
131            self.peek_stale_served.fetch_add(1, Ordering::Relaxed);
132        }
133        Ok(entry)
134    }
135
136    async fn invalidate(&self, key: &str) -> Result<(), CamelError> {
137        self.inner.invalidate(key).await;
138        self.invalidations.fetch_add(1, Ordering::Relaxed);
139        Ok(())
140    }
141
142    async fn clear(&self) -> Result<(), CamelError> {
143        self.inner.invalidate_all();
144        Ok(())
145    }
146
147    async fn stats(&self) -> CacheStats {
148        self.stats_snapshot()
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    fn entry() -> CacheEntry {
157        CacheEntry {
158            bytes: vec![1, 2, 3],
159            content_type: camel_api::cache::ContentType::Bytes,
160            expires_at: None,
161        }
162    }
163
164    #[tokio::test]
165    async fn get_returns_none_on_miss_some_on_hit() {
166        let repo = MemoryCacheRepository::new("test", 100);
167
168        repo.set("k", entry(), Some(Duration::from_secs(3600)))
169            .await
170            .unwrap();
171        let found = repo.get("k").await.unwrap();
172        assert!(found.is_some());
173        assert_eq!(found.unwrap().bytes, vec![1, 2, 3]);
174        assert_eq!(repo.get("absent").await.unwrap(), None);
175    }
176
177    #[tokio::test]
178    async fn get_returns_none_after_expiry_peek_stale_returns_entry() {
179        let repo = MemoryCacheRepository::new("test", 100);
180
181        repo.set("k", entry(), Some(Duration::from_millis(1)))
182            .await
183            .unwrap();
184        tokio::time::sleep(Duration::from_millis(10)).await;
185        assert_eq!(repo.get("k").await.unwrap(), None);
186        let stale = repo.peek_stale("k").await.unwrap();
187        assert!(stale.is_some());
188        assert_eq!(stale.unwrap().bytes, vec![1, 2, 3]);
189    }
190
191    #[tokio::test]
192    async fn set_with_none_ttl_stores_without_expiry() {
193        let repo = MemoryCacheRepository::new("test", 100);
194
195        repo.set("k", entry(), None).await.unwrap();
196        let found = repo.get("k").await.unwrap();
197        assert!(found.is_some());
198        assert!(found.unwrap().expires_at.is_none());
199    }
200
201    #[tokio::test]
202    async fn invalidate_is_noop_on_absent_key() {
203        let repo = MemoryCacheRepository::new("test", 100);
204        repo.invalidate("absent").await.unwrap();
205    }
206
207    #[tokio::test]
208    async fn max_capacity_bounds_entry_count() {
209        let repo = MemoryCacheRepository::new("test", 2);
210
211        repo.set("a", entry(), Some(Duration::from_secs(3600)))
212            .await
213            .unwrap();
214        repo.set("b", entry(), Some(Duration::from_secs(3600)))
215            .await
216            .unwrap();
217        repo.set("c", entry(), Some(Duration::from_secs(3600)))
218            .await
219            .unwrap();
220        repo.inner.run_pending_tasks().await;
221        assert!(repo.inner.entry_count() <= 2);
222    }
223
224    #[tokio::test]
225    async fn stats_reflects_hits_misses_evictions_entries() {
226        let repo = MemoryCacheRepository::new("test", 100);
227
228        repo.set("k", entry(), Some(Duration::from_secs(3600)))
229            .await
230            .unwrap();
231        repo.get("k").await.unwrap(); // hit
232        repo.get("absent").await.unwrap(); // miss
233        repo.inner.run_pending_tasks().await;
234
235        let stats = repo.stats().await;
236        assert_eq!(stats.hits, 1);
237        assert_eq!(stats.misses, 1);
238        assert!(stats.entries >= 1, "entries was {}", stats.entries);
239    }
240
241    #[tokio::test]
242    async fn stats_reports_peek_stale_served() {
243        let repo = MemoryCacheRepository::new("test", 100);
244
245        repo.set("k", entry(), Some(Duration::from_millis(1)))
246            .await
247            .unwrap();
248        tokio::time::sleep(Duration::from_millis(10)).await;
249        assert_eq!(repo.get("k").await.unwrap(), None); // miss/expired
250        assert!(repo.peek_stale("k").await.unwrap().is_some());
251        assert_eq!(repo.stats().await.peek_stale_served, 1);
252    }
253
254    #[tokio::test]
255    async fn stats_reports_invalidations_per_operation() {
256        let repo = MemoryCacheRepository::new("test", 100);
257
258        repo.set("a", entry(), None).await.unwrap();
259        repo.invalidate("a").await.unwrap();
260        repo.invalidate("absent").await.unwrap();
261        assert_eq!(repo.stats().await.invalidations, 2);
262    }
263
264    #[tokio::test]
265    async fn clear_empties_repository() {
266        let repo = MemoryCacheRepository::new("test", 100);
267
268        repo.set("a", entry(), None).await.unwrap();
269        repo.set("b", entry(), None).await.unwrap();
270        repo.clear().await.unwrap();
271        assert_eq!(repo.get("a").await.unwrap(), None);
272        assert_eq!(repo.get("b").await.unwrap(), None);
273    }
274
275    #[tokio::test]
276    async fn evictions_incremented_on_size_pressure() {
277        let repo = MemoryCacheRepository::new("test", 1);
278
279        repo.set("a", entry(), None).await.unwrap();
280        repo.set("b", entry(), None).await.unwrap();
281        repo.inner.run_pending_tasks().await;
282        assert!(repo.stats().await.evictions >= 1);
283    }
284
285    #[tokio::test]
286    async fn invalidate_prefix_on_memory_fails_naming_backend() {
287        let repo = MemoryCacheRepository::new("test", 100);
288        let err = repo.invalidate_prefix("ns:").await.unwrap_err();
289        assert!(
290            format!("{err}").contains("test"),
291            "error must name the backend, got: {err}"
292        );
293    }
294}
camel_core/cache/memory.rs

camel_core/cache/
memory.rs
