Skip to main content

linera_cache/
value_cache.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! A concurrent cache with efficient eviction and single-allocation guarantees.
5//!
6//! Values are stored internally as `Arc<V>`, so cache hits return a cheap
7//! `Arc` clone instead of cloning the underlying data. A secondary weak
8//! index (`papaya::HashMap<K, Weak<V>>`) ensures that at most one allocation
9//! exists per key: if the bounded cache evicts an entry while a consumer
10//! still holds an `Arc`, re-requesting the same key returns the same
11//! allocation instead of creating a duplicate.
12
13#[cfg(with_metrics)]
14use std::any::type_name;
15use std::{
16    borrow::Cow,
17    hash::Hash,
18    sync::{Arc, Weak},
19};
20
21use linera_base::{crypto::CryptoHash, hashed::Hashed};
22use papaya::{Compute, Operation};
23use quick_cache::sync::Cache;
24
25/// Default interval between dead-entry cleanup sweeps of the weak index.
26pub const DEFAULT_CLEANUP_INTERVAL_SECS: u64 = 30;
27
28/// A concurrent cache with efficient eviction and single-allocation guarantees.
29///
30/// Backed by `quick_cache` (S3-FIFO eviction) for bounded hot-path caching, plus
31/// a lock-free `papaya::HashMap` weak index for deduplication. Together they
32/// guarantee that at most one `Arc<V>` allocation exists per key at any time.
33///
34/// A background task periodically sweeps dead `Weak` entries from the index
35/// to prevent unbounded memory growth.
36pub struct ValueCache<K, V> {
37    cache: Cache<K, crate::Arc<V>>,
38    weak_index: Arc<papaya::HashMap<K, Weak<V>>>,
39}
40
41impl<K, V> ValueCache<K, V>
42where
43    K: Hash + Eq + Clone + Send + Sync + 'static,
44    V: Send + Sync + 'static,
45{
46    /// Creates a new `ValueCache` with the given bounded-cache capacity
47    /// and cleanup interval for the weak-reference index.
48    #[cfg(not(web))]
49    pub fn new(size: usize, cleanup_interval_secs: u64) -> Self {
50        let weak_index = Arc::new(papaya::HashMap::new());
51        Self::spawn_cleanup_task(
52            Arc::clone(&weak_index),
53            std::time::Duration::from_secs(cleanup_interval_secs),
54        );
55        ValueCache {
56            cache: Cache::new(size),
57            weak_index,
58        }
59    }
60
61    /// Creates a new `ValueCache` (web variant, no background cleanup task).
62    #[cfg(web)]
63    pub fn new(size: usize, _cleanup_interval_secs: u64) -> Self {
64        ValueCache {
65            cache: Cache::new(size),
66            weak_index: Arc::new(papaya::HashMap::new()),
67        }
68    }
69
70    /// Inserts a value into the cache, returning the canonical [`crate::Arc`].
71    ///
72    /// The value is wrapped in `Arc` internally. If a live `Arc` for this key
73    /// already exists (held by another consumer), the existing allocation is
74    /// reused and the new value is dropped.
75    pub fn insert(&self, key: &K, value: V) -> crate::Arc<V> {
76        self.dedup_insert(key, crate::Arc(Arc::new(value)))
77    }
78
79    /// Removes a value from the bounded cache.
80    ///
81    /// The weak index entry is intentionally kept — another consumer may
82    /// still hold an `Arc` to this value, and the weak index must be able
83    /// to deduplicate against it. Dead weak entries are cleaned up by the
84    /// background task.
85    pub fn remove(&self, key: &K) -> Option<crate::Arc<V>> {
86        let value = self.cache.peek(key);
87        if value.is_some() {
88            self.cache.remove(key);
89        }
90        Self::track_cache_usage(value)
91    }
92
93    /// Returns an [`crate::Arc`] to the value, checking both the bounded
94    /// cache and the weak index.
95    pub fn get(&self, key: &K) -> Option<crate::Arc<V>> {
96        // Tier 1: bounded cache (hot path)
97        if let Some(arc) = self.cache.get(key) {
98            return Self::track_cache_usage(Some(arc));
99        }
100
101        // Tier 2: weak index (catches evicted-but-still-held entries)
102        let guard = self.weak_index.guard();
103        if let Some(weak) = self.weak_index.get(key, &guard) {
104            if let Some(arc) = weak.upgrade() {
105                let arc = crate::Arc(arc);
106                // Re-insert into bounded cache for future fast lookups
107                self.cache.insert(key.clone(), arc.clone());
108                return Self::track_cache_usage(Some(arc));
109            }
110        }
111
112        Self::track_cache_usage(None)
113    }
114
115    /// Returns `true` if the value exists in either the bounded cache or
116    /// the weak index (with a live allocation).
117    pub fn contains(&self, key: &K) -> bool {
118        if self.cache.peek(key).is_some() {
119            return true;
120        }
121        let guard = self.weak_index.guard();
122        self.weak_index
123            .get(key, &guard)
124            .is_some_and(|weak| weak.strong_count() > 0)
125    }
126
127    /// Removes all dead `Weak` entries from the weak index.
128    #[cfg(with_testing)]
129    pub fn cleanup_dead_entries(&self) {
130        let guard = self.weak_index.guard();
131        self.weak_index
132            .retain(|_, weak| weak.strong_count() > 0, &guard);
133    }
134
135    /// Spawns a background task that periodically sweeps dead weak entries.
136    ///
137    /// No-op if no tokio runtime is available (e.g. in unit tests).
138    #[cfg(not(web))]
139    fn spawn_cleanup_task(
140        weak_index: Arc<papaya::HashMap<K, Weak<V>>>,
141        cleanup_interval: std::time::Duration,
142    ) {
143        if tokio::runtime::Handle::try_current().is_err() {
144            return;
145        }
146        tokio::spawn(async move {
147            let mut interval = tokio::time::interval(cleanup_interval);
148            loop {
149                interval.tick().await;
150                let guard = weak_index.guard();
151                weak_index.retain(|_, weak| weak.strong_count() > 0, &guard);
152            }
153        });
154    }
155
156    /// Core dedup logic: atomically checks the weak index for an existing
157    /// live allocation. If found, reuses it. Otherwise inserts the new Arc.
158    /// Returns the canonical `Arc`.
159    fn dedup_insert(&self, key: &K, new_arc: crate::Arc<V>) -> crate::Arc<V> {
160        let guard = self.weak_index.guard();
161        let weak = Arc::downgrade(&new_arc.0);
162
163        let result = self.weak_index.compute(
164            key.clone(),
165            |entry| match entry {
166                Some((_k, existing_weak)) => match existing_weak.upgrade() {
167                    Some(existing_arc) => Operation::Abort(existing_arc),
168                    None => Operation::Insert(weak.clone()),
169                },
170                None => Operation::Insert(weak.clone()),
171            },
172            &guard,
173        );
174
175        let canonical_arc = match result {
176            Compute::Inserted(..) | Compute::Updated { .. } => new_arc,
177            Compute::Aborted(existing_arc) => crate::Arc(existing_arc),
178            _ => unreachable!(),
179        };
180
181        self.cache.insert(key.clone(), canonical_arc.clone());
182        canonical_arc
183    }
184
185    fn track_cache_usage(maybe_value: Option<crate::Arc<V>>) -> Option<crate::Arc<V>> {
186        #[cfg(with_metrics)]
187        {
188            let metric = if maybe_value.is_some() {
189                &metrics::CACHE_HIT_COUNT
190            } else {
191                &metrics::CACHE_MISS_COUNT
192            };
193
194            metric
195                .with_label_values(&[type_name::<K>(), type_name::<V>()])
196                .inc();
197        }
198        maybe_value
199    }
200}
201
202impl<V: Clone + Send + Sync + 'static> ValueCache<CryptoHash, V> {
203    /// Inserts a value constructed from a [`Hashed<T>`] into the cache, keyed
204    /// by its hash, returning the canonical `Arc<V>`.
205    ///
206    /// The `value` is wrapped in a [`Cow`] so that it is only cloned if it
207    /// needs to be inserted in the cache.
208    pub fn insert_hashed<T>(&self, value: Cow<Hashed<T>>) -> crate::Arc<V>
209    where
210        T: Clone,
211        V: From<Hashed<T>>,
212    {
213        let hash = (*value).hash();
214        // Fast path: already in bounded cache
215        if let Some(arc) = self.cache.peek(&hash) {
216            return arc;
217        }
218        // Check weak index before cloning from Cow
219        let guard = self.weak_index.guard();
220        if let Some(weak) = self.weak_index.get(&hash, &guard) {
221            if let Some(arc) = weak.upgrade() {
222                let arc = crate::Arc(arc);
223                self.cache.insert(hash, arc.clone());
224                return arc;
225            }
226        }
227        drop(guard);
228        self.dedup_insert(&hash, crate::Arc(Arc::new(value.into_owned().into())))
229    }
230
231    /// Inserts multiple values constructed from [`Hashed<T>`]s into the cache.
232    #[cfg(with_testing)]
233    pub fn insert_all_hashed<'a, T>(&self, values: impl IntoIterator<Item = Cow<'a, Hashed<T>>>)
234    where
235        T: Clone + 'a,
236        V: From<Hashed<T>>,
237    {
238        for value in values {
239            self.insert_hashed(value);
240        }
241    }
242}
243
244#[cfg(with_metrics)]
245mod metrics {
246    use std::sync::LazyLock;
247
248    use linera_base::prometheus_util::register_int_counter_vec;
249    use prometheus::IntCounterVec;
250
251    pub static CACHE_HIT_COUNT: LazyLock<IntCounterVec> = LazyLock::new(|| {
252        register_int_counter_vec(
253            "value_cache_hit",
254            "Cache hits in `ValueCache`",
255            &["key_type", "value_type"],
256        )
257    });
258
259    pub static CACHE_MISS_COUNT: LazyLock<IntCounterVec> = LazyLock::new(|| {
260        register_int_counter_vec(
261            "value_cache_miss",
262            "Cache misses in `ValueCache`",
263            &["key_type", "value_type"],
264        )
265    });
266}
267
268#[cfg(test)]
269mod tests {
270    use std::borrow::Cow;
271
272    use linera_base::{crypto::CryptoHash, hashed::Hashed};
273    use serde::{Deserialize, Serialize};
274
275    use super::{ValueCache, DEFAULT_CLEANUP_INTERVAL_SECS};
276    use crate::Arc as CacheArc;
277
278    /// Test cache size for unit tests.
279    const TEST_CACHE_SIZE: usize = 10;
280
281    /// A minimal hashable value for testing `ValueCache<CryptoHash, Hashed<T>>`.
282    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
283    struct TestValue(u64);
284
285    impl linera_base::crypto::BcsHashable<'_> for TestValue {}
286
287    impl From<Hashed<TestValue>> for TestValue {
288        fn from(value: Hashed<TestValue>) -> Self {
289            value.into_inner()
290        }
291    }
292
293    fn create_test_value(n: u64) -> Hashed<TestValue> {
294        Hashed::new(TestValue(n))
295    }
296
297    fn create_test_values(iter: impl IntoIterator<Item = u64>) -> Vec<Hashed<TestValue>> {
298        iter.into_iter().map(create_test_value).collect()
299    }
300
301    fn new_hashed_cache(size: usize) -> ValueCache<CryptoHash, TestValue> {
302        ValueCache::new(size, DEFAULT_CLEANUP_INTERVAL_SECS)
303    }
304
305    fn new_string_cache(size: usize) -> ValueCache<u64, String> {
306        ValueCache::new(size, DEFAULT_CLEANUP_INTERVAL_SECS)
307    }
308
309    #[test]
310    fn test_retrieve_missing_value() {
311        let cache = new_hashed_cache(TEST_CACHE_SIZE);
312        let hash = CryptoHash::test_hash("Missing value");
313
314        assert!(cache.get(&hash).is_none());
315        assert!(!cache.contains(&hash));
316    }
317
318    #[test]
319    fn test_insert_and_get() {
320        let cache = new_hashed_cache(TEST_CACHE_SIZE);
321        let value = create_test_value(0);
322        let hash = value.hash();
323
324        cache.insert_hashed(Cow::Borrowed(&value));
325        assert!(cache.contains(&hash));
326        assert_eq!(cache.get(&hash).as_deref(), Some(value.inner()));
327    }
328
329    #[test]
330    fn test_insert_many_values() {
331        let cache = new_hashed_cache(TEST_CACHE_SIZE);
332        let values = create_test_values(0..TEST_CACHE_SIZE as u64);
333
334        for value in &values {
335            cache.insert_hashed(Cow::Borrowed(value));
336        }
337
338        for value in &values {
339            assert!(cache.contains(&value.hash()));
340            assert_eq!(cache.get(&value.hash()).as_deref(), Some(value.inner()));
341        }
342
343        // Batch insert
344        let cache2 = new_hashed_cache(TEST_CACHE_SIZE);
345        cache2.insert_all_hashed(values.iter().map(Cow::Borrowed));
346        for value in &values {
347            assert_eq!(cache2.get(&value.hash()).as_deref(), Some(value.inner()));
348        }
349    }
350
351    #[test]
352    fn test_reinsertion_dedup() {
353        let cache = new_hashed_cache(TEST_CACHE_SIZE);
354        let values = create_test_values(0..TEST_CACHE_SIZE as u64);
355
356        // First insert
357        let first_arcs: Vec<_> = values
358            .iter()
359            .map(|v| cache.insert_hashed(Cow::Borrowed(v)))
360            .collect();
361
362        // Re-inserting should return the same Arc (dedup)
363        for (value, first_arc) in values.iter().zip(&first_arcs) {
364            let second_arc = cache.insert_hashed(Cow::Borrowed(value));
365            assert!(CacheArc::ptr_eq(&second_arc, first_arc));
366        }
367    }
368
369    #[test]
370    fn test_eviction() {
371        let cache = new_hashed_cache(TEST_CACHE_SIZE);
372        let total = TEST_CACHE_SIZE * 3;
373        let values = create_test_values(0..total as u64);
374
375        for value in &values {
376            cache.insert_hashed(Cow::Borrowed(value));
377        }
378
379        let present_count = values.iter().filter(|v| cache.contains(&v.hash())).count();
380        assert!(
381            present_count <= TEST_CACHE_SIZE + 1,
382            "cache should not hold significantly more than its capacity, \
383             but has {present_count} entries for capacity {TEST_CACHE_SIZE}"
384        );
385        assert!(present_count > 0, "cache should still hold some entries");
386    }
387
388    #[test]
389    fn test_accessed_entry_survives_eviction() {
390        let cache = new_hashed_cache(TEST_CACHE_SIZE);
391        let promoted = create_test_value(0);
392        let promoted_hash = promoted.hash();
393
394        cache.insert_hashed(Cow::Borrowed(&promoted));
395        cache.get(&promoted_hash); // mark as hot
396
397        let extras = create_test_values(1..=TEST_CACHE_SIZE as u64 * 2);
398        for value in &extras {
399            cache.insert_hashed(Cow::Borrowed(value));
400        }
401
402        assert!(
403            cache.contains(&promoted_hash),
404            "recently accessed entry should survive eviction"
405        );
406    }
407
408    #[test]
409    fn test_promotion_of_reinsertion() {
410        let cache = new_hashed_cache(TEST_CACHE_SIZE);
411        let promoted = create_test_value(0);
412        let promoted_hash = promoted.hash();
413
414        let first = cache.insert_hashed(Cow::Borrowed(&promoted));
415        let second = cache.insert_hashed(Cow::Borrowed(&promoted));
416        assert!(CacheArc::ptr_eq(&first, &second));
417
418        let extras = create_test_values(1..=TEST_CACHE_SIZE as u64 * 2);
419        for value in &extras {
420            cache.insert_hashed(Cow::Borrowed(value));
421        }
422
423        assert!(
424            cache.contains(&promoted_hash),
425            "re-inserted entry should survive eviction"
426        );
427    }
428
429    #[test]
430    fn test_weak_index_dedup_after_eviction() {
431        let cache = new_string_cache(2);
432
433        // Insert and hold onto the Arc
434        let held = cache.insert(&1, "hello".to_string());
435
436        // Force eviction by filling the cache
437        cache.insert(&2, "world".to_string());
438        cache.insert(&3, "foo".to_string());
439        cache.insert(&4, "bar".to_string());
440
441        // Weak index should find it via the held Arc
442        let retrieved = cache
443            .get(&1)
444            .expect("held Arc should keep entry findable via weak index");
445        assert!(
446            CacheArc::ptr_eq(&retrieved, &held),
447            "must return same allocation, not a duplicate"
448        );
449
450        // Re-inserting should also return the same Arc
451        let reinserted = cache.insert(&1, "replacement".to_string());
452        assert!(CacheArc::ptr_eq(&reinserted, &held));
453        assert_eq!(&*reinserted, "hello");
454    }
455
456    #[test]
457    fn test_remove_preserves_weak_for_held_arcs() {
458        let cache = new_string_cache(TEST_CACHE_SIZE);
459
460        let held = cache.insert(&1, "hello".to_string());
461
462        // remove() evicts from bounded cache but NOT the weak index
463        cache.remove(&1);
464
465        // Still findable via weak index since we hold an Arc
466        let retrieved = cache.get(&1).expect("weak index should find held Arc");
467        assert!(CacheArc::ptr_eq(&retrieved, &held));
468    }
469
470    #[test]
471    fn test_remove_without_holder() {
472        let cache = new_string_cache(TEST_CACHE_SIZE);
473
474        cache.insert(&1, "hello".to_string());
475
476        // remove() without anyone holding an Arc — weak entry becomes dead
477        cache.remove(&1);
478        assert!(!cache.contains(&1));
479        assert!(cache.get(&1).is_none());
480    }
481
482    #[test]
483    fn test_cleanup_dead_entries() {
484        let cache = new_string_cache(2);
485
486        cache.insert(&1, "alive".to_string());
487        let _held = cache.get(&1).expect("just inserted"); // keep alive
488
489        cache.insert(&2, "dead".to_string());
490        // Don't hold key 2
491
492        // Force eviction of both by filling the cache
493        cache.insert(&3, "a".to_string());
494        cache.insert(&4, "b".to_string());
495        cache.insert(&5, "c".to_string());
496
497        cache.cleanup_dead_entries();
498
499        // Key 1 still findable (we hold an Arc)
500        assert!(cache.contains(&1));
501    }
502}