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}
30
31impl std::fmt::Debug for MemoryCacheRepository {
32    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        f.debug_struct("MemoryCacheRepository")
34            .field("name", &self.name)
35            .field("stats", &self.stats())
36            .finish()
37    }
38}
39
40impl MemoryCacheRepository {
41    /// Create a new memory-backed cache repository.
42    ///
43    /// `max_capacity` is the maximum number of entries before eviction.
44    pub fn new(name: impl Into<String>, max_capacity: usize) -> Self {
45        let hits = Arc::new(AtomicU64::new(0));
46        let misses = Arc::new(AtomicU64::new(0));
47        let evictions = Arc::new(AtomicU64::new(0));
48        let evictions_clone = Arc::clone(&evictions);
49
50        let inner = moka::future::CacheBuilder::new(max_capacity as u64)
51            .eviction_listener(move |_key, _value, cause| {
52                // Only count capacity-driven size evictions — NOT explicit invalidate() calls.
53                if cause == moka::notification::RemovalCause::Size {
54                    evictions_clone.fetch_add(1, Ordering::Relaxed);
55                }
56            })
57            .build();
58
59        Self {
60            name: name.into(),
61            inner,
62            hits,
63            misses,
64            evictions,
65        }
66    }
67}
68
69#[async_trait]
70impl CacheRepository for MemoryCacheRepository {
71    fn name(&self) -> &str {
72        &self.name
73    }
74
75    async fn get(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
76        match self.inner.get(key).await {
77            Some(entry) => {
78                let expired = entry
79                    .expires_at
80                    .map(|e| e <= SystemTime::now())
81                    .unwrap_or(false);
82                if expired {
83                    self.misses.fetch_add(1, Ordering::Relaxed);
84                    Ok(None)
85                } else {
86                    self.hits.fetch_add(1, Ordering::Relaxed);
87                    Ok(Some(entry))
88                }
89            }
90            None => {
91                self.misses.fetch_add(1, Ordering::Relaxed);
92                Ok(None)
93            }
94        }
95    }
96
97    async fn set(
98        &self,
99        key: &str,
100        mut value: CacheEntry,
101        ttl: Option<Duration>,
102    ) -> Result<(), CamelError> {
103        value.expires_at = ttl.map(|d| SystemTime::now() + d);
104        self.inner.insert(key.to_string(), value).await;
105        Ok(())
106    }
107
108    async fn peek_stale(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
109        Ok(self.inner.get(key).await)
110    }
111
112    async fn invalidate(&self, key: &str) -> Result<(), CamelError> {
113        self.inner.invalidate(key).await;
114        Ok(())
115    }
116
117    async fn clear(&self) -> Result<(), CamelError> {
118        self.inner.invalidate_all();
119        Ok(())
120    }
121
122    fn stats(&self) -> CacheStats {
123        CacheStats {
124            hits: self.hits.load(Ordering::Relaxed),
125            misses: self.misses.load(Ordering::Relaxed),
126            evictions: self.evictions.load(Ordering::Relaxed),
127            entries: self.inner.entry_count(),
128        }
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    fn entry() -> CacheEntry {
137        CacheEntry {
138            bytes: vec![1, 2, 3],
139            content_type: camel_api::cache::ContentType::Bytes,
140            expires_at: None,
141        }
142    }
143
144    #[tokio::test]
145    async fn get_returns_none_on_miss_some_on_hit() {
146        let repo = MemoryCacheRepository::new("test", 100);
147
148        repo.set("k", entry(), Some(Duration::from_secs(3600)))
149            .await
150            .unwrap();
151        let found = repo.get("k").await.unwrap();
152        assert!(found.is_some());
153        assert_eq!(found.unwrap().bytes, vec![1, 2, 3]);
154        assert_eq!(repo.get("absent").await.unwrap(), None);
155    }
156
157    #[tokio::test]
158    async fn get_returns_none_after_expiry_peek_stale_returns_entry() {
159        let repo = MemoryCacheRepository::new("test", 100);
160
161        repo.set("k", entry(), Some(Duration::from_millis(1)))
162            .await
163            .unwrap();
164        tokio::time::sleep(Duration::from_millis(10)).await;
165        assert_eq!(repo.get("k").await.unwrap(), None);
166        let stale = repo.peek_stale("k").await.unwrap();
167        assert!(stale.is_some());
168        assert_eq!(stale.unwrap().bytes, vec![1, 2, 3]);
169    }
170
171    #[tokio::test]
172    async fn set_with_none_ttl_stores_without_expiry() {
173        let repo = MemoryCacheRepository::new("test", 100);
174
175        repo.set("k", entry(), None).await.unwrap();
176        let found = repo.get("k").await.unwrap();
177        assert!(found.is_some());
178        assert!(found.unwrap().expires_at.is_none());
179    }
180
181    #[tokio::test]
182    async fn invalidate_is_noop_on_absent_key() {
183        let repo = MemoryCacheRepository::new("test", 100);
184        repo.invalidate("absent").await.unwrap();
185    }
186
187    #[tokio::test]
188    async fn max_capacity_bounds_entry_count() {
189        let repo = MemoryCacheRepository::new("test", 2);
190
191        repo.set("a", entry(), Some(Duration::from_secs(3600)))
192            .await
193            .unwrap();
194        repo.set("b", entry(), Some(Duration::from_secs(3600)))
195            .await
196            .unwrap();
197        repo.set("c", entry(), Some(Duration::from_secs(3600)))
198            .await
199            .unwrap();
200        repo.inner.run_pending_tasks().await;
201        assert!(repo.inner.entry_count() <= 2);
202    }
203
204    #[tokio::test]
205    async fn stats_reflects_hits_misses_evictions_entries() {
206        let repo = MemoryCacheRepository::new("test", 100);
207
208        repo.set("k", entry(), Some(Duration::from_secs(3600)))
209            .await
210            .unwrap();
211        repo.get("k").await.unwrap(); // hit
212        repo.get("absent").await.unwrap(); // miss
213        repo.inner.run_pending_tasks().await;
214
215        let stats = repo.stats();
216        assert_eq!(stats.hits, 1);
217        assert_eq!(stats.misses, 1);
218        assert!(stats.entries >= 1, "entries was {}", stats.entries);
219    }
220
221    #[tokio::test]
222    async fn clear_empties_repository() {
223        let repo = MemoryCacheRepository::new("test", 100);
224
225        repo.set("a", entry(), None).await.unwrap();
226        repo.set("b", entry(), None).await.unwrap();
227        repo.clear().await.unwrap();
228        assert_eq!(repo.get("a").await.unwrap(), None);
229        assert_eq!(repo.get("b").await.unwrap(), None);
230    }
231
232    #[tokio::test]
233    async fn evictions_incremented_on_size_pressure() {
234        let repo = MemoryCacheRepository::new("test", 1);
235
236        repo.set("a", entry(), None).await.unwrap();
237        repo.set("b", entry(), None).await.unwrap();
238        repo.inner.run_pending_tasks().await;
239        assert!(repo.stats().evictions >= 1);
240    }
241}
camel_core/cache/memory.rs

camel_core/cache/
memory.rs
