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 peek_row_silent(&self, key: &str) -> Result<Option<CacheEntry>, CamelError> {
137        // Raw store read: no hit/miss/peek accounting (trait contract).
138        Ok(self.inner.get(key).await)
139    }
140
141    async fn invalidate(&self, key: &str) -> Result<(), CamelError> {
142        self.inner.invalidate(key).await;
143        self.invalidations.fetch_add(1, Ordering::Relaxed);
144        Ok(())
145    }
146
147    async fn clear(&self) -> Result<(), CamelError> {
148        self.inner.invalidate_all();
149        // moka removal is eventual: without draining pending maintenance
150        // tasks, `entry_count` can still report evicted entries right after
151        // clear returns (bd rc-cdnz). Draining keeps clear observably
152        // synchronous for stats snapshots.
153        self.inner.run_pending_tasks().await;
154        Ok(())
155    }
156
157    async fn stats(&self) -> CacheStats {
158        self.stats_snapshot()
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165
166    fn entry() -> CacheEntry {
167        CacheEntry {
168            bytes: vec![1, 2, 3],
169            payload_path: None,
170            content_type: camel_api::cache::ContentType::Bytes,
171            expires_at: None,
172        }
173    }
174
175    #[tokio::test]
176    async fn get_returns_none_on_miss_some_on_hit() {
177        let repo = MemoryCacheRepository::new("test", 100);
178
179        repo.set("k", entry(), Some(Duration::from_secs(3600)))
180            .await
181            .unwrap();
182        let found = repo.get("k").await.unwrap();
183        assert!(found.is_some());
184        assert_eq!(found.unwrap().bytes, vec![1, 2, 3]);
185        assert_eq!(repo.get("absent").await.unwrap(), None);
186    }
187
188    #[tokio::test]
189    async fn get_returns_none_after_expiry_peek_stale_returns_entry() {
190        let repo = MemoryCacheRepository::new("test", 100);
191
192        repo.set("k", entry(), Some(Duration::from_millis(1)))
193            .await
194            .unwrap();
195        tokio::time::sleep(Duration::from_millis(10)).await;
196        assert_eq!(repo.get("k").await.unwrap(), None);
197        let stale = repo.peek_stale("k").await.unwrap();
198        assert!(stale.is_some());
199        assert_eq!(stale.unwrap().bytes, vec![1, 2, 3]);
200    }
201
202    #[tokio::test]
203    async fn set_with_none_ttl_stores_without_expiry() {
204        let repo = MemoryCacheRepository::new("test", 100);
205
206        repo.set("k", entry(), None).await.unwrap();
207        let found = repo.get("k").await.unwrap();
208        assert!(found.is_some());
209        assert!(found.unwrap().expires_at.is_none());
210    }
211
212    #[tokio::test]
213    async fn invalidate_is_noop_on_absent_key() {
214        let repo = MemoryCacheRepository::new("test", 100);
215        repo.invalidate("absent").await.unwrap();
216    }
217
218    #[tokio::test]
219    async fn clear_leaves_entry_count_immediately_zero() {
220        // Regression guard (bd rc-cdnz): moka removal is eventual, so clear
221        // drains pending maintenance tasks before returning. Loop to catch
222        // any scheduling jitter the drain might miss.
223        let repo = MemoryCacheRepository::new("test", 100);
224        for round in 0..20 {
225            for i in 0..5 {
226                repo.set(&format!("k{round}-{i}"), entry(), None)
227                    .await
228                    .unwrap();
229            }
230            repo.clear().await.unwrap();
231            assert_eq!(
232                repo.stats().await.entries,
233                0,
234                "entry count must be zero right after clear (round {round})"
235            );
236        }
237    }
238
239    #[tokio::test]
240    async fn max_capacity_bounds_entry_count() {
241        let repo = MemoryCacheRepository::new("test", 2);
242
243        repo.set("a", entry(), Some(Duration::from_secs(3600)))
244            .await
245            .unwrap();
246        repo.set("b", entry(), Some(Duration::from_secs(3600)))
247            .await
248            .unwrap();
249        repo.set("c", entry(), Some(Duration::from_secs(3600)))
250            .await
251            .unwrap();
252        repo.inner.run_pending_tasks().await;
253        assert!(repo.inner.entry_count() <= 2);
254    }
255
256    #[tokio::test]
257    async fn stats_reflects_hits_misses_evictions_entries() {
258        let repo = MemoryCacheRepository::new("test", 100);
259
260        repo.set("k", entry(), Some(Duration::from_secs(3600)))
261            .await
262            .unwrap();
263        repo.get("k").await.unwrap(); // hit
264        repo.get("absent").await.unwrap(); // miss
265        repo.inner.run_pending_tasks().await;
266
267        let stats = repo.stats().await;
268        assert_eq!(stats.hits, 1);
269        assert_eq!(stats.misses, 1);
270        assert!(stats.entries >= 1, "entries was {}", stats.entries);
271    }
272
273    #[tokio::test]
274    async fn stats_reports_peek_stale_served() {
275        let repo = MemoryCacheRepository::new("test", 100);
276
277        repo.set("k", entry(), Some(Duration::from_millis(1)))
278            .await
279            .unwrap();
280        tokio::time::sleep(Duration::from_millis(10)).await;
281        assert_eq!(repo.get("k").await.unwrap(), None); // miss/expired
282        assert!(repo.peek_stale("k").await.unwrap().is_some());
283        assert_eq!(repo.stats().await.peek_stale_served, 1);
284    }
285
286    #[tokio::test]
287    async fn stats_reports_invalidations_per_operation() {
288        let repo = MemoryCacheRepository::new("test", 100);
289
290        repo.set("a", entry(), None).await.unwrap();
291        repo.invalidate("a").await.unwrap();
292        repo.invalidate("absent").await.unwrap();
293        assert_eq!(repo.stats().await.invalidations, 2);
294    }
295
296    #[tokio::test]
297    async fn clear_empties_repository() {
298        let repo = MemoryCacheRepository::new("test", 100);
299
300        repo.set("a", entry(), None).await.unwrap();
301        repo.set("b", entry(), None).await.unwrap();
302        repo.clear().await.unwrap();
303        assert_eq!(repo.get("a").await.unwrap(), None);
304        assert_eq!(repo.get("b").await.unwrap(), None);
305    }
306
307    #[tokio::test]
308    async fn evictions_incremented_on_size_pressure() {
309        let repo = MemoryCacheRepository::new("test", 1);
310
311        repo.set("a", entry(), None).await.unwrap();
312        repo.set("b", entry(), None).await.unwrap();
313        repo.inner.run_pending_tasks().await;
314        assert!(repo.stats().await.evictions >= 1);
315    }
316
317    #[tokio::test]
318    async fn invalidate_prefix_on_memory_fails_naming_backend() {
319        let repo = MemoryCacheRepository::new("test", 100);
320        let err = repo.invalidate_prefix("ns:").await.unwrap_err();
321        assert!(
322            format!("{err}").contains("test"),
323            "error must name the backend, got: {err}"
324        );
325    }
326}
camel_core/cache/memory.rs

camel_core/cache/
memory.rs
