fix(store): deterministic LFU cache eviction (tie-break by key)

graph/src/util/lfu_cache.rs

@@ -54,9 +54,13 @@ impl<K: CacheWeight, V: Default + CacheWeight> CacheEntry<K, V> {
     }
 }
 
-// The priorities are `(stale, frequency)` tuples, first all stale entries will be popped and
-// then non-stale entries by least frequency.
-type Priority = (bool, Reverse<u64>);
+// The priorities are `(stale, frequency, key)` triples. First all stale entries will be popped
+// and then non-stale entries by least frequency. Ties on the first two components are broken by
+// the key so that eviction is deterministic and independent of the order in which entries were
+// inserted; without the key tie-breaker the `PriorityQueue` would break ties by heap position,
+// which depends on insertion order and therefore makes the eviction set differ between otherwise
+// identical runs. See https://github.com/graphprotocol/graph-node/issues/6706
+type Priority<K> = (bool, Reverse<u64>, Reverse<K>);
 
 /// Statistics about what happened during cache eviction
 pub struct EvictStats {
@@ -96,7 +100,7 @@ impl EvictStats {
 /// evictions entities are checked for staleness.
 #[derive(Debug)]
 pub struct LfuCache<K: Eq + Hash, V> {
-    queue: PriorityQueue<CacheEntry<K, V>, Priority>,
+    queue: PriorityQueue<CacheEntry<K, V>, Priority<K>>,
     total_weight: usize,
     stale_counter: u64,
     dead_weight: bool,
@@ -135,14 +139,17 @@ impl<K: Clone + Ord + Eq + Hash + Debug + CacheWeight, V: CacheWeight + Default>
         match self.get_mut(key.clone()) {
             None => {
                 self.total_weight += weight;
+                // Clone the key once more to break eviction-priority ties
+                // deterministically; see `type Priority<K>`.
+                let tie_break_key = key.clone();
                 self.queue.push(
                     CacheEntry {
                         weight,
                         key,
                         value,
                         will_stale: false,
                     },
-                    (false, Reverse(1)),
+                    (false, Reverse(1), Reverse(tie_break_key)),
                 );
             }
             Some(entry) => {
@@ -168,7 +175,7 @@ impl<K: Clone + Ord + Eq + Hash + Debug + CacheWeight, V: CacheWeight + Default>
         // Increment the frequency by 1
         let key_entry = CacheEntry::cache_key(key);
         self.queue
-            .change_priority_by(&key_entry, |(_, Reverse(f))| {
+            .change_priority_by(&key_entry, |(_, Reverse(f), _)| {
                 *f += 1;
             });
         self.accesses += 1;
@@ -194,7 +201,7 @@ impl<K: Clone + Ord + Eq + Hash + Debug + CacheWeight, V: CacheWeight + Default>
         // the absolute minimum and popping.
         let key_entry = CacheEntry::cache_key(key.clone());
         self.queue
-            .change_priority(&key_entry, (true, Reverse(u64::MIN)))
+            .change_priority(&key_entry, (true, Reverse(u64::MIN), Reverse(key.clone())))
             .and_then(|_| {
                 self.queue.pop().map(|(e, _)| {
                     assert_eq!(e.key, key_entry.key);
@@ -306,16 +313,16 @@ impl<K: Clone + Ord + Eq + Hash + Debug + CacheWeight, V: CacheWeight + Default>
 }
 
 impl<K: Ord + Eq + Hash + 'static, V: 'static> IntoIterator for LfuCache<K, V> {
-    type Item = (CacheEntry<K, V>, Priority);
+    type Item = (CacheEntry<K, V>, Priority<K>);
     type IntoIter = Box<dyn Iterator<Item = Self::Item>>;
 
     fn into_iter(self) -> Self::IntoIter {
         Box::new(self.queue.into_iter())
     }
 }
 
-impl<K: Ord + Eq + Hash, V> Extend<(CacheEntry<K, V>, Priority)> for LfuCache<K, V> {
-    fn extend<T: IntoIterator<Item = (CacheEntry<K, V>, Priority)>>(&mut self, iter: T) {
+impl<K: Ord + Eq + Hash, V> Extend<(CacheEntry<K, V>, Priority<K>)> for LfuCache<K, V> {
+    fn extend<T: IntoIterator<Item = (CacheEntry<K, V>, Priority<K>)>>(&mut self, iter: T) {
         self.queue.extend(iter);
     }
 }

graph/tests/lfu_cache_determinism.rs

@@ -0,0 +1,50 @@
+//! Regression test for https://github.com/graphprotocol/graph-node/issues/6706
+//!
+//! Eviction ties on the `(stale, frequency)` priority are the normal case, not
+//! the exception. Which of the tied entries is evicted must not depend on the
+//! order in which they were inserted: a caller can insert entries in an order
+//! that is nondeterministic per process (e.g. when iterating a `HashMap` whose
+//! iteration order is randomized per process), and that would otherwise make
+//! the eviction set — and therefore cache hit rates / store read counts —
+//! differ between otherwise identical runs.
+
+use graph::prelude::CacheWeight;
+use graph::util::lfu_cache::LfuCache;
+
+#[derive(Default, Debug, PartialEq, Eq)]
+struct Weight(usize);
+
+impl CacheWeight for Weight {
+    fn weight(&self) -> usize {
+        self.indirect_weight()
+    }
+
+    fn indirect_weight(&self) -> usize {
+        self.0
+    }
+}
+
+/// Insert the same six entries, each of equal weight and frequency 1, so every
+/// eviction candidate is tied on the priority, evict down to a max weight that
+/// leaves two entries, and return the surviving keys.
+fn survivors_in_order(order: &[&str]) -> Vec<String> {
+    let mut cache: LfuCache<String, Weight> = LfuCache::new();
+    for &key in order {
+        cache.insert(key.to_string(), Weight(1));
+    }
+    cache.evict(6);
+    let mut survivors: Vec<String> = cache.iter().map(|(k, _)| k.clone()).collect();
+    survivors.sort();
+    survivors
+}
+
+#[test]
+fn eviction_is_independent_of_insertion_order() {
+    let order_a = ["k1", "k2", "k3", "k4", "k5", "k6"];
+    let order_b = ["k6", "k5", "k4", "k3", "k2", "k1"];
+    let order_c = ["k3", "k6", "k2", "k5", "k1", "k4"];
+
+    let reference = survivors_in_order(&order_a);
+    assert_eq!(survivors_in_order(&order_b), reference);
+    assert_eq!(survivors_in_order(&order_c), reference);
+}