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    /// Stable instance name distinguishing this cache in metrics; several
38    /// caches may share the same key/value types within one process.
39    ///
40    /// Must be unique among the `ValueCache` instances of a process:
41    /// instances sharing a name and key/value types report into the same
42    /// metric series, so their gauges would overwrite each other. The name
43    /// is used verbatim as the `cache` Prometheus label value; any UTF-8
44    /// string is valid, but prefer a short, stable `snake_case` identifier
45    /// since dashboards and alerts query it.
46    name: &'static str,
47    cache: Arc<Cache<K, crate::Arc<V>>>,
48    weak_index: Arc<papaya::HashMap<K, Weak<V>>>,
49}
50
51impl<K, V> ValueCache<K, V>
52where
53    K: Hash + Eq + Clone + Send + Sync + 'static,
54    V: Send + Sync + 'static,
55{
56    /// Creates a new `ValueCache` with the given instance name (used as the
57    /// `cache` metric label), bounded-cache capacity, and cleanup interval
58    /// for the weak-reference index.
59    #[cfg(not(web))]
60    pub fn new(name: &'static str, size: usize, cleanup_interval_secs: u64) -> Self {
61        let cache = Arc::new(Cache::new(size));
62        let weak_index = Arc::new(papaya::HashMap::new());
63        // Report quick_cache's actual capacity, which may round the requested
64        // size up, so that occupancy (entries / capacity) cannot exceed 1.
65        #[cfg(with_metrics)]
66        metrics::CACHE_CAPACITY
67            .with_label_values(&[name, type_name::<K>(), type_name::<V>()])
68            .set(i64::try_from(cache.capacity()).unwrap_or(i64::MAX));
69        Self::spawn_cleanup_task(
70            Arc::clone(&weak_index),
71            #[cfg(with_metrics)]
72            (name, Arc::clone(&cache)),
73            std::time::Duration::from_secs(cleanup_interval_secs),
74        );
75        ValueCache {
76            name,
77            cache,
78            weak_index,
79        }
80    }
81
82    /// Creates a new `ValueCache` (web variant, no background cleanup task).
83    #[cfg(web)]
84    pub fn new(name: &'static str, size: usize, _cleanup_interval_secs: u64) -> Self {
85        ValueCache {
86            name,
87            cache: Arc::new(Cache::new(size)),
88            weak_index: Arc::new(papaya::HashMap::new()),
89        }
90    }
91
92    /// Returns the instance name of this cache.
93    pub fn name(&self) -> &'static str {
94        self.name
95    }
96
97    /// Returns the number of entries currently held in the bounded cache.
98    ///
99    /// This is the live occupancy (excludes the weak dedup index), and is
100    /// periodically sampled as the `value_cache_entries` gauge for the
101    /// occupancy-vs-capacity diagnosis.
102    pub fn len(&self) -> usize {
103        self.cache.len()
104    }
105
106    /// Returns `true` if the bounded cache currently holds no entries.
107    pub fn is_empty(&self) -> bool {
108        self.cache.is_empty()
109    }
110
111    /// Inserts a value into the cache, returning the canonical [`crate::Arc`].
112    ///
113    /// The value is wrapped in `Arc` internally. If a live `Arc` for this key
114    /// already exists (held by another consumer), the existing allocation is
115    /// reused and the new value is dropped.
116    pub fn insert(&self, key: &K, value: V) -> crate::Arc<V> {
117        self.dedup_insert(key, crate::Arc(Arc::new(value)))
118    }
119
120    /// Removes a value from the bounded cache.
121    ///
122    /// The weak index entry is intentionally kept — another consumer may
123    /// still hold an `Arc` to this value, and the weak index must be able
124    /// to deduplicate against it. Dead weak entries are cleaned up by the
125    /// background task.
126    pub fn remove(&self, key: &K) -> Option<crate::Arc<V>> {
127        let value = self.cache.peek(key);
128        if value.is_some() {
129            self.cache.remove(key);
130        }
131        self.track_cache_usage(value)
132    }
133
134    /// Returns an [`crate::Arc`] to the value, checking both the bounded
135    /// cache and the weak index.
136    pub fn get(&self, key: &K) -> Option<crate::Arc<V>> {
137        // Tier 1: bounded cache (hot path)
138        if let Some(arc) = self.cache.get(key) {
139            return self.track_cache_usage(Some(arc));
140        }
141
142        // Tier 2: weak index (catches evicted-but-still-held entries)
143        let guard = self.weak_index.guard();
144        if let Some(weak) = self.weak_index.get(key, &guard) {
145            if let Some(arc) = weak.upgrade() {
146                let arc = crate::Arc(arc);
147                // Re-insert into bounded cache for future fast lookups
148                self.cache.insert(key.clone(), arc.clone());
149                return self.track_cache_usage(Some(arc));
150            }
151        }
152
153        self.track_cache_usage(None)
154    }
155
156    /// Returns `true` if the value exists in either the bounded cache or
157    /// the weak index (with a live allocation).
158    pub fn contains(&self, key: &K) -> bool {
159        if self.cache.peek(key).is_some() {
160            return true;
161        }
162        let guard = self.weak_index.guard();
163        self.weak_index
164            .get(key, &guard)
165            .is_some_and(|weak| weak.strong_count() > 0)
166    }
167
168    /// Removes all dead `Weak` entries from the weak index.
169    #[cfg(with_testing)]
170    pub fn cleanup_dead_entries(&self) {
171        let guard = self.weak_index.guard();
172        self.weak_index
173            .retain(|_, weak| weak.strong_count() > 0, &guard);
174    }
175
176    /// Spawns a background task that periodically sweeps dead weak entries.
177    ///
178    /// No-op if no tokio runtime is available (e.g. in unit tests).
179    #[cfg(not(web))]
180    fn spawn_cleanup_task(
181        weak_index: Arc<papaya::HashMap<K, Weak<V>>>,
182        #[cfg(with_metrics)] metrics_handle: (&'static str, Arc<Cache<K, crate::Arc<V>>>),
183        cleanup_interval: std::time::Duration,
184    ) {
185        if tokio::runtime::Handle::try_current().is_err() {
186            return;
187        }
188        tokio::spawn(async move {
189            let mut interval = tokio::time::interval(cleanup_interval);
190            loop {
191                interval.tick().await;
192                {
193                    let guard = weak_index.guard();
194                    weak_index.retain(|_, weak| weak.strong_count() > 0, &guard);
195                }
196                // Refresh the occupancy gauge on the same cadence as the sweep
197                // (after the guard is dropped); keeps it off the cache hot path.
198                #[cfg(with_metrics)]
199                {
200                    let (name, cache) = &metrics_handle;
201                    metrics::CACHE_ENTRIES
202                        .with_label_values(&[name, type_name::<K>(), type_name::<V>()])
203                        .set(i64::try_from(cache.len()).unwrap_or(i64::MAX));
204                }
205            }
206        });
207    }
208
209    /// Core dedup logic: atomically checks the weak index for an existing
210    /// live allocation. If found, reuses it. Otherwise inserts the new Arc.
211    /// Returns the canonical `Arc`.
212    fn dedup_insert(&self, key: &K, new_arc: crate::Arc<V>) -> crate::Arc<V> {
213        let guard = self.weak_index.guard();
214        let weak = Arc::downgrade(&new_arc.0);
215
216        let result = self.weak_index.compute(
217            key.clone(),
218            |entry| match entry {
219                Some((_k, existing_weak)) => match existing_weak.upgrade() {
220                    Some(existing_arc) => Operation::Abort(existing_arc),
221                    None => Operation::Insert(weak.clone()),
222                },
223                None => Operation::Insert(weak.clone()),
224            },
225            &guard,
226        );
227
228        let canonical_arc = match result {
229            Compute::Inserted(..) | Compute::Updated { .. } => new_arc,
230            Compute::Aborted(existing_arc) => crate::Arc(existing_arc),
231            _ => unreachable!(),
232        };
233
234        self.cache.insert(key.clone(), canonical_arc.clone());
235        canonical_arc
236    }
237
238    fn track_cache_usage(&self, maybe_value: Option<crate::Arc<V>>) -> Option<crate::Arc<V>> {
239        #[cfg(with_metrics)]
240        {
241            let metric = if maybe_value.is_some() {
242                &metrics::CACHE_HIT_COUNT
243            } else {
244                &metrics::CACHE_MISS_COUNT
245            };
246
247            metric
248                .with_label_values(&[self.name, type_name::<K>(), type_name::<V>()])
249                .inc();
250        }
251        maybe_value
252    }
253}
254
255impl<V: Clone + Send + Sync + 'static> ValueCache<CryptoHash, V> {
256    /// Inserts a value constructed from a [`Hashed<T>`] into the cache, keyed
257    /// by its hash, returning the canonical `Arc<V>`.
258    ///
259    /// The `value` is wrapped in a [`Cow`] so that it is only cloned if it
260    /// needs to be inserted in the cache.
261    pub fn insert_hashed<T>(&self, value: Cow<Hashed<T>>) -> crate::Arc<V>
262    where
263        T: Clone,
264        V: From<Hashed<T>>,
265    {
266        let hash = (*value).hash();
267        // Fast path: already in bounded cache
268        if let Some(arc) = self.cache.peek(&hash) {
269            return arc;
270        }
271        // Check weak index before cloning from Cow
272        let guard = self.weak_index.guard();
273        if let Some(weak) = self.weak_index.get(&hash, &guard) {
274            if let Some(arc) = weak.upgrade() {
275                let arc = crate::Arc(arc);
276                self.cache.insert(hash, arc.clone());
277                return arc;
278            }
279        }
280        drop(guard);
281        self.dedup_insert(&hash, crate::Arc(Arc::new(value.into_owned().into())))
282    }
283
284    /// Inserts multiple values constructed from [`Hashed<T>`]s into the cache.
285    #[cfg(with_testing)]
286    pub fn insert_all_hashed<'a, T>(&self, values: impl IntoIterator<Item = Cow<'a, Hashed<T>>>)
287    where
288        T: Clone + 'a,
289        V: From<Hashed<T>>,
290    {
291        for value in values {
292            self.insert_hashed(value);
293        }
294    }
295}
296
297#[cfg(with_metrics)]
298pub(crate) mod metrics {
299    use linera_base::prometheus_util::{register_int_counter_vec, register_int_gauge_vec};
300    use prometheus::{IntCounterVec, IntGaugeVec};
301
302    /// Shared label set: `cache` is the per-instance name passed to
303    /// [`super::ValueCache::new`], required because several caches may share
304    /// the same key/value types within one process.
305    const LABELS: &[&str] = &["cache", "key_type", "value_type"];
306
307    linera_base::declare_metrics! {
308        pub static CACHE_HIT_COUNT: IntCounterVec =
309            register_int_counter_vec("value_cache_hit", "Cache hits in `ValueCache`", LABELS);
310
311        pub static CACHE_MISS_COUNT: IntCounterVec =
312            register_int_counter_vec("value_cache_miss", "Cache misses in `ValueCache`", LABELS);
313
314        pub static CACHE_ENTRIES: IntGaugeVec =
315            register_int_gauge_vec(
316                "value_cache_entries",
317                "Number of entries held in the bounded `ValueCache`, sampled at \
318                 each cleanup sweep",
319                LABELS,
320            );
321
322        pub static CACHE_CAPACITY: IntGaugeVec =
323            register_int_gauge_vec(
324                "value_cache_capacity",
325                "Maximum number of entries of the bounded `ValueCache`",
326                LABELS,
327            );
328    }
329}
330
331#[cfg(test)]
332mod tests {
333    use std::borrow::Cow;
334
335    use linera_base::{crypto::CryptoHash, hashed::Hashed};
336    use serde::{Deserialize, Serialize};
337
338    use super::{ValueCache, DEFAULT_CLEANUP_INTERVAL_SECS};
339    use crate::Arc as CacheArc;
340
341    /// Test cache size for unit tests.
342    const TEST_CACHE_SIZE: usize = 10;
343
344    /// A minimal hashable value for testing `ValueCache<CryptoHash, Hashed<T>>`.
345    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
346    struct TestValue(u64);
347
348    impl linera_base::crypto::BcsHashable<'_> for TestValue {}
349
350    impl From<Hashed<TestValue>> for TestValue {
351        fn from(value: Hashed<TestValue>) -> Self {
352            value.into_inner()
353        }
354    }
355
356    fn create_test_value(n: u64) -> Hashed<TestValue> {
357        Hashed::new(TestValue(n))
358    }
359
360    fn create_test_values(iter: impl IntoIterator<Item = u64>) -> Vec<Hashed<TestValue>> {
361        iter.into_iter().map(create_test_value).collect()
362    }
363
364    fn new_hashed_cache(size: usize) -> ValueCache<CryptoHash, TestValue> {
365        ValueCache::new("test_hashed", size, DEFAULT_CLEANUP_INTERVAL_SECS)
366    }
367
368    fn new_string_cache(size: usize) -> ValueCache<u64, String> {
369        ValueCache::new("test_string", size, DEFAULT_CLEANUP_INTERVAL_SECS)
370    }
371
372    #[test]
373    fn test_retrieve_missing_value() {
374        let cache = new_hashed_cache(TEST_CACHE_SIZE);
375        let hash = CryptoHash::test_hash("Missing value");
376
377        assert!(cache.get(&hash).is_none());
378        assert!(!cache.contains(&hash));
379    }
380
381    #[test]
382    fn test_insert_and_get() {
383        let cache = new_hashed_cache(TEST_CACHE_SIZE);
384        let value = create_test_value(0);
385        let hash = value.hash();
386
387        cache.insert_hashed(Cow::Borrowed(&value));
388        assert!(cache.contains(&hash));
389        assert_eq!(cache.get(&hash).as_deref(), Some(value.inner()));
390    }
391
392    #[test]
393    fn test_insert_many_values() {
394        let cache = new_hashed_cache(TEST_CACHE_SIZE);
395        let values = create_test_values(0..TEST_CACHE_SIZE as u64);
396
397        for value in &values {
398            cache.insert_hashed(Cow::Borrowed(value));
399        }
400
401        for value in &values {
402            assert!(cache.contains(&value.hash()));
403            assert_eq!(cache.get(&value.hash()).as_deref(), Some(value.inner()));
404        }
405
406        // Batch insert
407        let cache2 = new_hashed_cache(TEST_CACHE_SIZE);
408        cache2.insert_all_hashed(values.iter().map(Cow::Borrowed));
409        for value in &values {
410            assert_eq!(cache2.get(&value.hash()).as_deref(), Some(value.inner()));
411        }
412    }
413
414    #[test]
415    fn test_reinsertion_dedup() {
416        let cache = new_hashed_cache(TEST_CACHE_SIZE);
417        let values = create_test_values(0..TEST_CACHE_SIZE as u64);
418
419        // First insert
420        let first_arcs: Vec<_> = values
421            .iter()
422            .map(|v| cache.insert_hashed(Cow::Borrowed(v)))
423            .collect();
424
425        // Re-inserting should return the same Arc (dedup)
426        for (value, first_arc) in values.iter().zip(&first_arcs) {
427            let second_arc = cache.insert_hashed(Cow::Borrowed(value));
428            assert!(CacheArc::ptr_eq(&second_arc, first_arc));
429        }
430    }
431
432    #[test]
433    fn test_eviction() {
434        let cache = new_hashed_cache(TEST_CACHE_SIZE);
435        let total = TEST_CACHE_SIZE * 3;
436        let values = create_test_values(0..total as u64);
437
438        for value in &values {
439            cache.insert_hashed(Cow::Borrowed(value));
440        }
441
442        let present_count = values.iter().filter(|v| cache.contains(&v.hash())).count();
443        assert!(
444            present_count <= TEST_CACHE_SIZE + 1,
445            "cache should not hold significantly more than its capacity, \
446             but has {present_count} entries for capacity {TEST_CACHE_SIZE}"
447        );
448        assert!(present_count > 0, "cache should still hold some entries");
449    }
450
451    #[test]
452    fn test_len_and_is_empty() {
453        let cache = new_string_cache(TEST_CACHE_SIZE);
454        assert!(cache.is_empty());
455        assert_eq!(cache.len(), 0);
456
457        cache.insert(&1, "a".to_string());
458        cache.insert(&2, "b".to_string());
459        assert!(!cache.is_empty());
460        assert_eq!(cache.len(), 2);
461
462        // Filling well beyond capacity caps the bounded-cache occupancy, which
463        // is exactly what the `value_cache_entries` gauge reports. Note that
464        // `len()` counts only the bounded cache, not the weak dedup index.
465        for i in 3..(TEST_CACHE_SIZE as u64 * 3) {
466            cache.insert(&i, format!("v{i}"));
467        }
468        assert!(
469            cache.len() <= TEST_CACHE_SIZE,
470            "bounded-cache occupancy {} must not exceed capacity {TEST_CACHE_SIZE}",
471            cache.len(),
472        );
473    }
474
475    #[test]
476    fn test_accessed_entry_survives_eviction() {
477        let cache = new_hashed_cache(TEST_CACHE_SIZE);
478        let promoted = create_test_value(0);
479        let promoted_hash = promoted.hash();
480
481        cache.insert_hashed(Cow::Borrowed(&promoted));
482        cache.get(&promoted_hash); // mark as hot
483
484        let extras = create_test_values(1..=TEST_CACHE_SIZE as u64 * 2);
485        for value in &extras {
486            cache.insert_hashed(Cow::Borrowed(value));
487        }
488
489        assert!(
490            cache.contains(&promoted_hash),
491            "recently accessed entry should survive eviction"
492        );
493    }
494
495    #[test]
496    fn test_promotion_of_reinsertion() {
497        let cache = new_hashed_cache(TEST_CACHE_SIZE);
498        let promoted = create_test_value(0);
499        let promoted_hash = promoted.hash();
500
501        let first = cache.insert_hashed(Cow::Borrowed(&promoted));
502        let second = cache.insert_hashed(Cow::Borrowed(&promoted));
503        assert!(CacheArc::ptr_eq(&first, &second));
504
505        let extras = create_test_values(1..=TEST_CACHE_SIZE as u64 * 2);
506        for value in &extras {
507            cache.insert_hashed(Cow::Borrowed(value));
508        }
509
510        assert!(
511            cache.contains(&promoted_hash),
512            "re-inserted entry should survive eviction"
513        );
514    }
515
516    #[test]
517    fn test_weak_index_dedup_after_eviction() {
518        let cache = new_string_cache(2);
519
520        // Insert and hold onto the Arc
521        let held = cache.insert(&1, "hello".to_string());
522
523        // Force eviction by filling the cache
524        cache.insert(&2, "world".to_string());
525        cache.insert(&3, "foo".to_string());
526        cache.insert(&4, "bar".to_string());
527
528        // Weak index should find it via the held Arc
529        let retrieved = cache
530            .get(&1)
531            .expect("held Arc should keep entry findable via weak index");
532        assert!(
533            CacheArc::ptr_eq(&retrieved, &held),
534            "must return same allocation, not a duplicate"
535        );
536
537        // Re-inserting should also return the same Arc
538        let reinserted = cache.insert(&1, "replacement".to_string());
539        assert!(CacheArc::ptr_eq(&reinserted, &held));
540        assert_eq!(&*reinserted, "hello");
541    }
542
543    #[test]
544    fn test_remove_preserves_weak_for_held_arcs() {
545        let cache = new_string_cache(TEST_CACHE_SIZE);
546
547        let held = cache.insert(&1, "hello".to_string());
548
549        // remove() evicts from bounded cache but NOT the weak index
550        cache.remove(&1);
551
552        // Still findable via weak index since we hold an Arc
553        let retrieved = cache.get(&1).expect("weak index should find held Arc");
554        assert!(CacheArc::ptr_eq(&retrieved, &held));
555    }
556
557    #[test]
558    fn test_remove_without_holder() {
559        let cache = new_string_cache(TEST_CACHE_SIZE);
560
561        cache.insert(&1, "hello".to_string());
562
563        // remove() without anyone holding an Arc — weak entry becomes dead
564        cache.remove(&1);
565        assert!(!cache.contains(&1));
566        assert!(cache.get(&1).is_none());
567    }
568
569    #[test]
570    fn test_cleanup_dead_entries() {
571        let cache = new_string_cache(2);
572
573        cache.insert(&1, "alive".to_string());
574        let _held = cache.get(&1).expect("just inserted"); // keep alive
575
576        cache.insert(&2, "dead".to_string());
577        // Don't hold key 2
578
579        // Force eviction of both by filling the cache
580        cache.insert(&3, "a".to_string());
581        cache.insert(&4, "b".to_string());
582        cache.insert(&5, "c".to_string());
583
584        cache.cleanup_dead_entries();
585
586        // Key 1 still findable (we hold an Arc)
587        assert!(cache.contains(&1));
588    }
589}
linera_cache/value_cache.rs

linera_cache/
value_cache.rs
