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

camel_core/cache/
memory.rs
