rushdb_storage/
graph_cache.rs

1//! Resident adjacency of one relationship type, cached per snapshot.
2//!
3//! A traversal reads each record's adjacency rows with one range scan per
4//! record and direction. When traversals at one snapshot keep reading the
5//! same relationships, holding them in memory is cheaper: the graph a type
6//! resolves to is a function of the snapshot's commit index, so it is kept
7//! under `(key, commit index)` and shared by every reader of that snapshot,
8//! like the owner-set cache. A commit moves readers to a new index; older
9//! graphs age out by LRU.
10//!
11//! **When to build** is a rent-or-buy decision. Building reads every
12//! adjacency row of the tenant once (the rows are keyed by record, not by
13//! type). Traversals report the rows they read through storage for a key at
14//! a commit index ([`GraphCache::rent`]); once that reaches the cost of one
15//! build, the next traversal builds and keeps the graph. A read-mostly
16//! tenant pays at most twice the build; a tenant whose commits keep moving
17//! the snapshot never builds, because its rent restarts at every commit.
18//!
19//! Charged to its own budget (`graph-cache-bytes`) under the managed limit;
20//! a graph that does not fit is simply not kept, and traversals read
21//! storage as before.
22
23use std::any::Any;
24use std::collections::HashMap;
25use std::sync::{Arc, Mutex};
26
27use crate::cache::LogicalCache;
28use crate::resources::ByteBudget;
29use crate::StorageError;
30
31/// A cached graph: opaque to storage (the query layer's adjacency type).
32pub type Graph = Arc<dyn Any + Send + Sync>;
33
34/// The store's per-snapshot adjacency cache.
35pub struct GraphCache {
36    cache: Mutex<Option<LogicalCache<(Vec<u8>, u64), Graph>>>,
37    /// Rows read through storage per key, at the commit index they were
38    /// read at.
39    rent: Mutex<HashMap<Vec<u8>, (u64, u64)>>,
40}
41
42impl GraphCache {
43    /// Build under `budget` (`graph-cache-bytes`); without one there is no
44    /// cache: every lookup misses and nothing is ever built.
45    ///
46    /// # Errors
47    /// Returns [`StorageError::InvalidConfiguration`] if the cache cannot
48    /// reserve its budget.
49    pub(crate) fn new(budget: Option<ByteBudget>) -> Result<Self, StorageError> {
50        let cache = budget
51            .map(LogicalCache::new)
52            .transpose()
53            .map_err(|e| StorageError::InvalidConfiguration(e.to_string()))?;
54        Ok(Self {
55            cache: Mutex::new(cache),
56            rent: Mutex::new(HashMap::new()),
57        })
58    }
59
60    /// Whether graphs can be kept at all.
61    #[must_use]
62    pub fn enabled(&self) -> bool {
63        self.cache.lock().is_ok_and(|cache| cache.is_some())
64    }
65
66    /// The graph cached for `key` at commit index `ci`.
67    #[must_use]
68    pub fn get(&self, key: &[u8], ci: u64) -> Option<Graph> {
69        let mut guard = self.cache.lock().ok()?;
70        guard.as_mut()?.get(&(key.to_vec(), ci)).cloned()
71    }
72
73    /// Keep `graph` (about `bytes` resident) for `key` at `ci`. Returns
74    /// whether it is now resident.
75    pub fn insert(&self, key: Vec<u8>, ci: u64, graph: Graph, bytes: usize) -> bool {
76        let Ok(mut guard) = self.cache.lock() else {
77            return false;
78        };
79        let Some(cache) = guard.as_mut() else {
80            return false;
81        };
82        // Graphs of older snapshots serve only readers still pinned there;
83        // the newest one is what later traversals use.
84        cache.retain(|(_, at), _| *at >= ci);
85        if let Ok(mut rent) = self.rent.lock() {
86            rent.remove(&key);
87        }
88        cache.insert((key, ci), graph, bytes)
89    }
90
91    /// Record `rows` adjacency rows read through storage for `key` at `ci`,
92    /// and whether the rows read at `ci` now pay for one build of
93    /// `build_rows` rows.
94    pub fn rent(&self, key: &[u8], ci: u64, rows: u64, build_rows: u64) -> bool {
95        if !self.enabled() {
96            return false;
97        }
98        let Ok(mut rent) = self.rent.lock() else {
99            return false;
100        };
101        // Rent paid at an older snapshot buys nothing at this one.
102        rent.retain(|_, (at, _)| *at >= ci);
103        let spent = rent.entry(key.to_vec()).or_insert((ci, 0));
104        if spent.0 != ci {
105            *spent = (ci, 0);
106        }
107        spent.1 = spent.1.saturating_add(rows);
108        spent.1 >= build_rows
109    }
110}
rushdb_storage/graph_cache.rs

rushdb_storage/
graph_cache.rs
