refactor: prefill RPC cache for `collect_events`#7077

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

hanabi1224 merged 18 commits into main from hm/prefill-cache-collect-events

May 21, 2026

CHANGELOG.md

Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@

### Breaking

- [#7073](https://github.com/ChainSafe/forest/pull/7073): Replaced the underlying cache engine across the node. The eviction policy is no longer strict LRU — it is now CLOCK-PRO via [`quick_cache`](https://crates.io/crates/quick_cache), which is scan-resistant and typically gives higher hit rates on chain workloads. The Prometheus metric names `lru_cache_hit_total` and `lru_cache_miss_total` are renamed to `cache_hit_total` and `cache_miss_total`. **Operators must update dashboards, alert rules, and recording rules** that reference the old names. Label set (`kind="..."`) is unchanged.
- [#7073](https://github.com/ChainSafe/forest/pull/7073) and [#7077](https://github.com/ChainSafe/forest/pull/7077): Replaced the underlying cache engine across the node. The eviction policy is no longer strict LRU — it is now CLOCK-PRO via [`quick_cache`](https://crates.io/crates/quick_cache), which is scan-resistant and typically gives higher hit rates on chain workloads. Refactored internal cache metrics to include `hits` and `misses` for all automatically. The old metrics `lru_cache_hit_total` and `lru_cache_miss_total` are deprecated in favor of `cache_{name}_hits` and `cache_{name}_misses`. Details can be found in https://forest-docs.pages.dev/reference/metrics

### Added

Expand Down

Cargo.toml

-Original file line number
+Diff line change
@@ Expand Up / @@ -180,6 +180,7 @@ prometheus-client = "0.23" @@
 puruspe = "0.4"
 quick-protobuf = "0.8"
 quick-protobuf-codec = "0.3"
+quick_cache = { version = "0.6", features = ["stats"] }
 rand = "0.8"
 rand_chacha = "0.3"
 rand_distr = "0.4"
@@ Expand DownExpand Up / @@ -239,7 +240,6 @@ zstd = "0.13" @@
 # optional dependencies
 console-subscriber = { version = "0.5", features = ["parking_lot"], optional = true }
-quick_cache = "0.6"
 sqlx = { version = "0.8", default-features = false, features = ["sqlite", "runtime-tokio", "macros"], optional = true }
 tikv-jemallocator = { version = "0.6", optional = true }
 tracing-loki = { version = "0.2", default-features = false, features = ["compat-0-2-1", "rustls"], optional = true }
@@ Expand Down @@

docs/docs/developers/deep_dives/chain_follower.md

-Original file line number
+Diff line change
@@ Expand Up @@
 When the `SyncStateMachine` receives a `SyncEvent::BadTipset` event, it takes two important actions to protect the node:
-. **Cache the Bad Block:** It adds the CID of every block in the failed tipset to the `BadBlockCache` struct. This is an LRU cache that prevents the node from wasting resources by re-fetching or re-validating a block that is already known to be invalid. (`src/chain_sync/bad_block_cache.rs`)
+. **Cache the Bad Block:** It adds the CID of every block in the failed tipset to the `BadBlockCache` struct. This is a `CLOCK-Pro` cache that prevents the node from wasting resources by re-fetching or re-validating a block that is already known to be invalid. (`src/chain_sync/bad_block_cache.rs`)
 . **Prune Descendants:** It traverses its internal map of tipsets and removes all known descendants of the bad tipset. Since a child of an invalid block is also invalid, this prunes entire invalid forks from the processing queue.
 ### Committing to the Chain
@@ Expand Down @@

docs/docs/users/guides/monitoring/metrics.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -11,14 +11,22 @@ curl localhost:6116/metrics @@
 Sample output:
 ```console
-# HELP lru_cache_miss Stats of lru cache miss.
-# TYPE lru_cache_miss counter
-lru_cache_miss_total{kind="tipset"} 7199
-lru_cache_miss_total{kind="sm_tipset"} 181
-# HELP lru_cache_hit Stats of lru cache hit.
-# TYPE lru_cache_hit counter
-lru_cache_hit_total{kind="sm_tipset"} 913
-lru_cache_hit_total{kind="tipset"} 971846
+# HELP cache_tipset_size_bytes Size of cache tipset in bytes
+# TYPE cache_tipset_size_bytes gauge
+# UNIT cache_tipset_size_bytes bytes
+cache_tipset_size_bytes 10877000
+# HELP cache_tipset_len Length of cache tipset
+# TYPE cache_tipset_len gauge
+cache_tipset_len 2880
+# HELP cache_tipset_cap Capacity of cache tipset
+# TYPE cache_tipset_cap gauge
+cache_tipset_cap 2880
+# HELP cache_tipset_hits Cache hits of tipset
+# TYPE cache_tipset_hits gauge
+cache_tipset_hits 19795
+# HELP cache_tipset_misses Cache misses of tipset
+# TYPE cache_tipset_misses gauge
+cache_tipset_misses 39026
 ...
 ```
@@ Expand Down @@

docs/docs/users/reference/metrics.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -9,8 +9,6 @@ title: Metrics @@
 | `libp2p_messsage_total` | Counter | Count | Total number of `libp2p` messages by type |
 | `invalid_tipset_total` | Counter | Count | Total number of invalid tipsets received over `gossipsub` |
 | `head_epoch` | Gauge | Epoch | Latest epoch synchronized to the node |
-| `lru_cache_hit` | Counter | Count | Stats of `lru` cache hit. Indexed by `kind` |
-| `lru_cache_miss` | Counter | Count | Stats of `lru` cache miss. Indexed by `kind` |
 | `rpc_method_failure` | Counter | Count | Number of failed RPC calls. Indexed by `method` |
 | `rpc_processing_time` | Histogram | Milliseconds | Duration of RPC method processing. Indexed by `method` |
 | `peer_failure_total` | Counter | Count | Total number of failed peer requests |
@@ Expand All / @@ -29,6 +27,11 @@ title: Metrics @@
 | `process_start_time_seconds` | Gauge | Seconds | Time that the process started (in seconds since the UNIX epoch) |
 | `process_uptime_seconds` | Counter | Seconds | Total time since the process started |
 | `libp2p_bandwidth_bytes_total` | Counter | Bytes | Bandwidth usage by direction and transport protocols. Indexed by `protocols` and `direction` |
+| `cache_{name}_size_bytes` | Gauge | Bytes | Cache size in bytes |
+| `cache_{name}_len` | Gauge | Count | Cache entry count |
+| `cache_{name}_cap` | Gauge | Count | Cache entry capacity |
+| `cache_{name}_hits` | Counter | Count | Cache hit count |
+| `cache_{name}_misses` | Counter | Count | Cache miss count |
 <details>
  <summary>Example `bitswap_message_count_total` output</summary>
@@ Expand DownExpand Up / @@ -74,16 +77,6 @@ bitswap_get_block_time_bucket{le="+Inf"} 0 @@
 ```
 </details>
-<details>
- <summary>Example `lru_cache_miss` output</summary>
-```
-# HELP lru_cache_miss Stats of lru cache miss.
-# TYPE lru_cache_miss counter
-lru_cache_miss_total{kind="sm_tipset"} 37
-lru_cache_miss_total{kind="tipset"} 7046
-```
-</details>
 <details>
  <summary>Example `rpc_processing_time` output</summary>
 ```
@@ Expand DownExpand Up / @@ -380,3 +373,25 @@ process_start_time_seconds 1742912218.100066 @@
 process_uptime_seconds_total 84.24605
 ```
 </details>
+<details>
+ <summary>Example `cache_{name}_{type}` output</summary>
+```
+# HELP cache_tipset_size_bytes Size of cache tipset in bytes
+# TYPE cache_tipset_size_bytes gauge
+# UNIT cache_tipset_size_bytes bytes
+cache_tipset_size_bytes 10877000
+# HELP cache_tipset_len Length of cache tipset
+# TYPE cache_tipset_len gauge
+cache_tipset_len 2880
+# HELP cache_tipset_cap Capacity of cache tipset
+# TYPE cache_tipset_cap gauge
+cache_tipset_cap 2880
+# HELP cache_tipset_hits Cache hits of tipset
+# TYPE cache_tipset_hits gauge
+cache_tipset_hits 19795
+# HELP cache_tipset_misses Cache misses of tipset
+# TYPE cache_tipset_misses gauge
+cache_tipset_misses 39026
+```
+</details>

src/beacon/drand.rs

Original file line numberDiff line numberDiff line change
@@ -246,7 +246,7 @@ impl DrandBeacon {
 }

 fn is_verified(&self, entry: &BeaconEntry) -> bool {
 self.verified_beacons.peek_cloned(&entry.round()).as_deref() == Some(entry)
 self.verified_beacons.get(&entry.round()).as_deref() == Some(entry)
 }
}

Expand DownExpand Up@@ -309,21 +309,21 @@ impl Beacon for DrandBeacon {
 };

 if is_valid && !validated.is_empty() {
 let cap = self.verified_beacons.cap();
 if cap < validated.len() {
 tracing::warn!(%cap, validated_len=%validated.len(), "verified_beacons.cap() is too small");
 let capacity = self.verified_beacons.capacity() as usize;
 if capacity < validated.len() {
 tracing::warn!(%capacity, validated_len=%validated.len(), "verified_beacons.capacity() is too small");
 }
 for entry in validated {
 self.verified_beacons
 .push(entry.round(), Arc::new(entry.clone()));
 .insert(entry.round(), Arc::new(entry.clone()));
 }
 }

 Ok(is_valid)
 }

 async fn entry(&self, round: u64) -> anyhow::Result<BeaconEntry> {
 let cached = self.verified_beacons.peek_cloned(&round);
 let cached = self.verified_beacons.get(&round);
 match cached {
 Some(cached_entry) => Ok(Arc::unwrap_or_clone(cached_entry)),
 None => {
Expand Down

src/beacon/signatures/mod.rs

Original file line numberDiff line numberDiff line change
@@ -16,10 +16,10 @@ pub use bls_signatures::{PublicKey as PublicKeyOnG1, Signature as SignatureOnG2}
const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_";
const CSUITE_G2: &[u8] = b"BLS_SIG_BLS12381G2_XMD:SHA-256_SSWU_RO_NUL_";

#[derive(Debug, Clone, Eq, PartialEq)]
#[derive(Debug, Clone, Eq, PartialEq, derive_more::Deref)]
pub struct PublicKeyOnG2(pub(crate) G2Projective);

#[derive(Debug, Clone, Eq, PartialEq)]
#[derive(Debug, Clone, Eq, PartialEq, derive_more::Deref)]
pub struct SignatureOnG1(pub(crate) G1Affine);

/// Ported from <https://docs.rs/bls-signatures/0.15.0/src/bls_signatures/signature.rs.html#214>
@@ -36,7 +36,7 @@ pub fn verify_messages_unchained(
 return true;
 }

 let public_key: G2Affine = public_key.as_affine();
 let public_key: G2Affine = public_key.to_affine();
 // zero key & single message should fail
 if n_messages == 1 && public_key.is_identity().into() {
 return false;
Expand Down

src/beacon/signatures/public_key_impls.rs

-Original file line number
+Diff line change
@@ Expand Up / @@ -5,10 +5,6 @@ use super::*; @@
 use zerocopy::FromBytes as _;
 impl PublicKeyOnG2 {
- pub fn as_affine(&self) -> G2Affine {
- self.0.to_affine()
- }
  pub fn verify(&self, message: impl AsRef<[u8]>, signature: &SignatureOnG1) -> bool {
  verify_messages_unchained(self, &[message.as_ref()], &[signature])
  }
@@ Expand All / @@ -27,7 +23,7 @@ impl PublicKeyOnG2 { @@
  }
  pub fn as_bytes(&self) -> [u8; G2Affine::compressed_size()] {
- self.0.to_affine().to_compressed()
+ self.to_affine().to_compressed()
  }
 }
@@ Expand Down @@

src/beacon/signatures/signature_impls.rs

-Original file line number
+Diff line change
@@ Expand Up / @@ -40,6 +40,6 @@ impl SignatureOnG1 { @@
  }
  pub fn as_bytes(&self) -> [u8; G1Affine::compressed_size()] {
- self.0.to_compressed()
+ self.to_compressed()
  }
 }

src/blocks/tipset.rs

-Original file line number
+Diff line change
@@ Expand Up / @@ -45,6 +45,7 @@ use thiserror::Error; @@
  Ord,
  GetSize,
  derive_more::IntoIterator,
+ derive_more::Deref,
 )]
 pub struct TipsetKey(#[into_iterator(owned, ref)] SmallCidNonEmptyVec);
@@ Expand All / @@ -63,11 +64,6 @@ impl TipsetKey { @@
  })
  }
- /// Returns `true` if the tipset key contains the given CID.
- pub fn contains(&self, cid: Cid) -> bool {
- self.0.contains(cid)
- }
  /// Returns a non-empty collection of `CID`
  pub fn into_cids(self) -> NonEmpty<Cid> {
  self.0.into_cids()
@@ Expand All / @@ -78,21 +74,6 @@ impl TipsetKey { @@
  self.0.clone().into_cids()
  }
- /// Returns an iterator of `CID`s.
- pub fn iter(&self) -> impl Iterator<Item = Cid> + '_ {
- self.0.iter()
- }
- /// Returns the number of `CID`s
- pub fn len(&self) -> usize {
- self.0.len()
- }
- // To suppress `#[warn(clippy::len_without_is_empty)]`
- pub fn is_empty(&self) -> bool {
- false
- }
  /// Terse representation of the tipset key.
  /// `bafy2bzaceaqrqoasufr7gdwrbhvlfy2xmc4e5sdzekjgyha2kldxigu73gilo`
  /// becomes `eaq...ilo`. The `bafy2bzac` prefix is removed.
@@ Expand Down @@

src/chain/store/chain_store.rs

-Original file line number
+Diff line change
@@ Expand Up / @@ -341,7 +341,13 @@ impl ChainStore { @@
  .messages_in_tipset_cache()
  .get_or_insert_with(ts.key(), || {
  let bmsgs = BlockMessages::for_tipset(self.db(), ts)?;
- Ok(bmsgs.into_iter().flat_map(|bm| bm.messages).collect_vec())
+ anyhow::Ok(
+ bmsgs
+ .into_iter()
+ .flat_map(|bm| bm.messages)
+ .collect_vec()
+ .into(),
+ )
  })?)
  }
@@ Expand DownExpand Up / @@ -589,47 +595,21 @@ where @@
 /// on performance measurements, is resource-intensive and can be a bottleneck for certain
 /// use-cases. This cache is intended to be used with a complementary function;
 /// [`messages_for_tipset_with_cache`].
-pub struct MessagesInTipsetCache {
- cache: SizeTrackingCache<TipsetKey, Arc<Vec<ChainMessage>>>,
-}
+#[derive(derive_more::Deref)]
+pub struct MessagesInTipsetCache(SizeTrackingCache<TipsetKey, Arc<Vec<ChainMessage>>>);
 impl MessagesInTipsetCache {
  pub fn new(capacity: NonZeroUsize) -> Self {
- Self {
- cache: SizeTrackingCache::new_with_metrics("msg_in_tipset", capacity),
- }
- }
- pub fn get(&self, key: &TipsetKey) -> Option<Arc<Vec<ChainMessage>>> {
- self.cache.get_cloned(key)
- }
- pub fn get_or_insert_with<F>(
- &self,
- key: &TipsetKey,
- f: F,
- ) -> anyhow::Result<Arc<Vec<ChainMessage>>>
- where
- F: FnOnce() -> anyhow::Result<Vec<ChainMessage>>,
- {
- if let Some(cached) = self.get(key) {
- Ok(cached)
- } else {
- Ok(self.insert(key.clone(), f()?))
- }
- }
- pub fn insert(&self, key: TipsetKey, mut value: Vec<ChainMessage>) -> Arc<Vec<ChainMessage>> {
- value.shrink_to_fit();
- let value = Arc::new(value);
- self.cache.push(key, value.clone());
- value
+ Self(SizeTrackingCache::new_with_metrics(
+ "msg_in_tipset",
+ capacity,
+ ))
  }
  /// Reads the intended cache size for this process from the environment or uses the default.
  fn read_cache_size() -> NonZeroUsize {
  // Arbitrary number, can be adjusted
- const DEFAULT: NonZeroUsize = nonzero!(1024usize);
+ const DEFAULT: NonZeroUsize = nonzero!(8192usize);
  std::env::var("FOREST_MESSAGES_IN_TIPSET_CACHE_SIZE")
  .ok()
  .and_then(|s| s.parse().ok())
@@ Expand All / @@ -645,9 +625,7 @@ impl Default for MessagesInTipsetCache { @@
 impl ShallowClone for MessagesInTipsetCache {
  fn shallow_clone(&self) -> Self {
- Self {
- cache: self.cache.shallow_clone(),
- }
+ Self(self.deref().shallow_clone())
  }
 }
@@ Expand DownExpand Up / @@ -736,20 +714,20 @@ mod tests { @@
  )]);
  assert!(cache.get(&key1).is_none());
- let msgs = vec![Message::default().into()];
+ let msgs = Arc::new(vec![Message::default().into()]);
  cache.insert(key1.clone(), msgs.clone());
- assert_eq!(&msgs, &*cache.get(&key1).unwrap());
+ assert_eq!(&msgs, &cache.get(&key1).unwrap());
  let inserter_executed: std::sync::atomic::AtomicBool =
  std::sync::atomic::AtomicBool::new(false);
  let key_inserter = || {
  inserter_executed.store(true, std::sync::atomic::Ordering::Relaxed);
- Ok(msgs.clone())
+anyhow::Ok(msgs.clone())
  };
  assert_eq!(
  &msgs,
- &*cache.get_or_insert_with(&key1, key_inserter).unwrap()
+ &cache.get_or_insert_with(&key1, key_inserter).unwrap()
  );
  assert!(!inserter_executed.load(std::sync::atomic::Ordering::Relaxed));
@@ Expand All / @@ -761,7 +739,7 @@ mod tests { @@
  assert!(cache.get(&key2).is_none());
  assert_eq!(
  &msgs,
- &*cache.get_or_insert_with(&key2, key_inserter).unwrap()
+ &cache.get_or_insert_with(&key2, key_inserter).unwrap()
  );
  assert!(inserter_executed.load(std::sync::atomic::Ordering::Relaxed));
  }
@@ Expand Down @@

Navigation Menu

Uh oh!

refactor: prefill RPC cache for `collect_events`#7077

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

refactor: prefill RPC cache for collect_events#7077

Uh oh!

refactor: prefill RPC cache for collect_events#7077

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

refactor: prefill RPC cache for `collect_events`#7077

refactor: prefill RPC cache for `collect_events`#7077