perf(keeper): replace account recency lru with lock-free s3-fifo

Cargo.toml

@@ -55,6 +55,7 @@ bytemuck = "1.25.2"
 cfg-if = "1.0.4"
 clonetree = "0.0.2"
 criterion = "0.8.2"
+crossbeam-queue = "0.3.14"
 derive_more = "2.1.1"
 env_logger = "0.11.11"
 fjall = { version = "3.1.10", default-features = false }

engine/tests/recency.rs

@@ -0,0 +1,91 @@
+//! Recency eviction under RPC fan-out: accounts read on every request must
+//! stay resident while each request also touches many one-off accounts.
+//!
+//! `missing_accounts` returns exactly the accounts a synchronizing caller
+//! fetches upstream, so its size per round is the upstream fetch cost. The
+//! test drains evictions the way the validator does: claim the cached eviction
+//! under the account lease and delete the account.
+#![cfg(test)]
+
+use engine::{
+    Engine,
+    testkit::{Pacing, TestEngine},
+};
+use keeper::testkit::{Dirs, keeper_builder};
+use solana_account::{AccountBuilder, AccountMode};
+use solana_pubkey::Pubkey;
+use tokio::sync::mpsc::UnboundedReceiver;
+
+const CAPACITY: usize = 256;
+const HOT: usize = 50;
+const ONE_OFF_PER_ROUND: usize = 300;
+const BATCH: usize = 100;
+const ROUNDS: usize = 15;
+/// Rounds allowed for hot accounts to graduate from probation. Each round's
+/// one-off traffic exceeds the cache, so a hot account evicted before its
+/// first hit reaches main only if its ghost entry survives the round (about
+/// two thirds of the time here). Stragglers shrink geometrically, so after
+/// this many rounds the chance any of the 50 remains is about 0.1%.
+const WARMUP_ROUNDS: usize = 12;
+
+/// Materializes every requested account the engine reports missing, then
+/// applies the resulting evictions. Returns how many accounts were fetched.
+async fn request(
+    engine: &Engine,
+    evictions: &mut UnboundedReceiver<Pubkey>,
+    keys: &[Pubkey],
+    owner: Pubkey,
+) -> usize {
+    let missing = engine.missing_accounts(keys).await.unwrap();
+    let fetched = missing.len();
+    for accessor in missing {
+        let account = AccountBuilder::default()
+            .lamports(2_000_000)
+            .owner(owner)
+            .mode(AccountMode::Mirrored)
+            .slot(1)
+            .data(vec![1]);
+        accessor.materialize(account, None).await.unwrap();
+    }
+    while let Ok(evicted) = evictions.try_recv() {
+        let accessor = engine.account(evicted).await.unwrap();
+        if let Some(accessor) = accessor.into_cached_eviction() {
+            accessor.delete().await.unwrap();
+        }
+    }
+    fetched
+}
+
+/// Proves hot accounts stop being re-fetched once they graduate from
+/// probation, while every request also fans out to one-off accounts that
+/// exceed the whole cache. An LRU re-fetches most of the hot set every round.
+#[tokio::test(flavor = "multi_thread")]
+async fn hot_accounts_survive_one_off_fanout() {
+    let dirs = Dirs::default();
+    let mut builder = keeper_builder(&dirs);
+    builder.accountsdb.lru_capacity = CAPACITY;
+    let te = TestEngine::from_builder(dirs, builder, Pacing::External).await;
+    let mut evictions = te.accounts().subscribe_evictions().unwrap();
+    let owner = Pubkey::new_unique();
+    let hot: Vec<_> = (0..HOT).map(|_| Pubkey::new_unique()).collect();
+
+    let mut refetched_per_round = Vec::with_capacity(ROUNDS);
+    for _ in 0..ROUNDS {
+        refetched_per_round.push(request(&te, &mut evictions, &hot, owner).await);
+        let one_off: Vec<_> = (0..ONE_OFF_PER_ROUND).map(|_| Pubkey::new_unique()).collect();
+        for batch in one_off.chunks(BATCH) {
+            assert_eq!(
+                request(&te, &mut evictions, batch, owner).await,
+                batch.len(),
+                "one-off accounts are always fetched"
+            );
+        }
+    }
+
+    println!("hot accounts fetched per round: {refetched_per_round:?}");
+    assert!(
+        refetched_per_round.iter().skip(WARMUP_ROUNDS).all(|count| *count == 0),
+        "hot accounts are still evicted by one-off traffic and re-fetched upstream \
+         after {WARMUP_ROUNDS} warm-up rounds: {refetched_per_round:?}"
+    );
+}

keeper/Cargo.toml

@@ -24,6 +24,7 @@ nucleus = { workspace = true, features = ["config", "notifier", "runtime"] }
 
 ahash = { workspace = true }
 arc-swap = { workspace = true }
+crossbeam-queue = { workspace = true }
 derive_more = { workspace = true, features = ["from"] }
 flume = { workspace = true }
 oneshot = { workspace = true }

keeper/src/accessor.rs

@@ -47,9 +47,12 @@ pub struct AccountsAccessor<'a> {
 }
 
 impl<'a> AccountsAccessor<'a> {
-    /// Promotes an existing recency entry without loading or pinning the account.
+    /// Records a read hit on a resident recency entry without acquiring a
+    /// lease or loading the account. Lock-free; absent accounts are ignored.
+    ///
+    /// Frequently touched accounts are retained over accounts read once.
     pub fn touch(&self, pubkey: &Pubkey) {
-        self.keeper.caches.accounts.lru.get_sync(pubkey);
+        self.keeper.caches.accounts.recency.touch(pubkey);
     }
 
     /// Waits for exclusive mutation ownership of `pubkey` and records the

keeper/src/cache.rs

@@ -10,13 +10,14 @@ use ledger::{
 };
 use nucleus::{Slot, ledger::Block, notifier::EventNotifier};
 use parking_lot::Mutex;
-use scc::{HashCache, HashMap, hash_cache::Entry as CacheEntry, hash_map::Entry};
+use scc::{HashMap, hash_map::Entry};
 use solana_account::AccountMode;
 use solana_hash::Hash as SolanaHash;
 use solana_pubkey::Pubkey;
 
 use crate::{
     metrics,
+    s3fifo::{Evicted, S3Fifo},
     subscriptions::{Subscription, Unicast},
 };
 
@@ -68,8 +69,8 @@ impl Caches {
 
 /// Account access cache along with per-account mutation reservations.
 pub(crate) struct AccountCache {
-    /// Accounts that do not require persistence, ordered by recent access.
-    pub(crate) lru: HashCache<Pubkey, (), RandomState>,
+    /// Resident accounts that do not require persistence, under S3-FIFO eviction.
+    pub(crate) recency: S3Fifo<Pubkey>,
     /// In-flight account mutations keyed by account pubkey.
     pub(crate) reservations: HashMap<Pubkey, Arc<EventNotifier>, RandomState>,
     /// Pubkeys evicted when a materialization displaces a cold account.
@@ -79,9 +80,8 @@ pub(crate) struct AccountCache {
 impl AccountCache {
     /// Creates mutation coordination with the requested recency capacity.
     pub(crate) fn new(capacity: usize) -> Self {
-        let lru = HashCache::with_capacity_and_hasher(capacity, capacity, Default::default());
         Self {
-            lru,
+            recency: S3Fifo::new(capacity),
             reservations: Default::default(),
             evictions: Unicast::new(Subscription::Evictions),
         }
@@ -137,7 +137,7 @@ impl AccountLease {
 
     /// Updates recency after materializing the account in `mode`.
     pub fn materialized(&mut self, mode: AccountMode) {
-        if let Some(pubkey) = self.cache.track(self.pubkey, mode) {
+        for pubkey in self.cache.track(self.pubkey, mode) {
             self.cache.evictions.send(|| pubkey);
         }
     }
@@ -150,7 +150,7 @@ impl AccountLease {
     /// Returns whether an earlier recency eviction still applies to the
     /// observed account and cache entry.
     pub fn cached_eviction_applies(&self) -> bool {
-        self.evictable() && !self.cache.lru.contains_sync(&self.pubkey)
+        self.evictable() && !self.cache.recency.contains(&self.pubkey)
     }
 }
 
@@ -163,40 +163,32 @@ impl Drop for AccountLease {
 }
 
 impl AccountCache {
-    /// Tracks a materialized account and returns one displaced from recency.
-    fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Option<Pubkey> {
+    /// Tracks a materialized account and returns the accounts it displaced.
+    fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Evicted<Pubkey> {
         if !mode.requires_persistence() {
             return self.admit_recency(pubkey);
         }
         self.remove_recency(&pubkey);
-        None
+        Evicted::new()
     }
 
-    /// Promotes an existing entry or admits `pubkey`, returning any displaced
-    /// account.
-    fn admit_recency(&self, pubkey: Pubkey) -> Option<Pubkey> {
-        let entry = match self.lru.entry_sync(pubkey) {
-            CacheEntry::Occupied(entry) => {
-                // Entry access does not promote recency; release its lock and
-                // use the cache access API for that operation.
-                drop(entry);
-                self.lru.get_sync(&pubkey);
-                return None;
-            }
-            CacheEntry::Vacant(entry) => entry,
-        };
-        let (evicted, _) = entry.put_entry(());
-        if evicted.is_some() {
-            metrics::account_cache_eviction();
-        } else {
+    /// Counts a hit on a resident entry or admits `pubkey`, returning any
+    /// displaced accounts.
+    fn admit_recency(&self, pubkey: Pubkey) -> Evicted<Pubkey> {
+        let admission = self.recency.admit(pubkey);
+        if admission.inserted {
             metrics::account_cache_insert();
         }
-        evicted.map(|entry| entry.0)
+        for _ in &admission.evicted {
+            metrics::account_cache_eviction();
+            metrics::account_cache_remove();
+        }
+        admission.evicted
     }
 
     /// Removes `pubkey` from recency and records an actual removal.
     fn remove_recency(&self, pubkey: &Pubkey) {
-        if self.lru.remove_sync(pubkey).is_some() {
+        if self.recency.remove(pubkey) {
             metrics::account_cache_remove();
         }
     }

keeper/src/lib.rs

@@ -62,6 +62,7 @@ pub mod builder;
 mod cache;
 pub mod error;
 mod metrics;
+mod s3fifo;
 mod subscriptions;
 mod util;