store: Thread cache_size to ChainStore

Pass the per-chain cache_size configuration from TOML config through
StoreBuilder and BlockStore to ChainStore, where it will be used
to determine which blocks should be treated as uncached.

Author: lutter

node/src/store_builder.rs

@@ -24,7 +24,8 @@ pub struct StoreBuilder {
     subscription_manager: Arc<SubscriptionManager>,
     chain_head_update_listener: Arc<PostgresChainHeadUpdateListener>,
     /// Map network names to the shards where they are/should be stored
-    chains: HashMap<String, ShardName>,
+    /// and their cache_size setting
+    chains: HashMap<String, (ShardName, i32)>,
     pub coord: Arc<PoolCoordinator>,
     registry: Arc<MetricsRegistry>,
 }
@@ -65,7 +66,7 @@ impl StoreBuilder {
         let chains = HashMap::from_iter(config.chains.chains.iter().map(|(name, chain)| {
             let shard = ShardName::new(chain.shard.to_string())
                 .expect("config validation catches invalid names");
-            (name.to_string(), shard)
+            (name.to_string(), (shard, chain.cache_size))
         }));
 
         let chain_head_update_listener = Arc::new(PostgresChainHeadUpdateListener::new(
@@ -177,15 +178,18 @@ impl StoreBuilder {
         logger: &Logger,
         pools: HashMap<ShardName, ConnectionPool>,
         subgraph_store: Arc<SubgraphStore>,
-        chains: HashMap<String, ShardName>,
+        chains: HashMap<String, (ShardName, i32)>,
         networks: Vec<String>,
         registry: Arc<MetricsRegistry>,
     ) -> Arc<DieselStore> {
         let networks = networks
             .into_iter()
             .map(|name| {
-                let shard = chains.get(&name).unwrap_or(&*PRIMARY_SHARD).clone();
-                (name, shard)
+                let (shard, cache_size) = chains
+                    .get(&name)
+                    .cloned()
+                    .unwrap_or_else(|| (PRIMARY_SHARD.clone(), 500));
+                (name, shard, cache_size)
             })
             .collect();
 

store/postgres/src/block_store.rs

@@ -218,7 +218,9 @@ pub struct Inner {
     /// previous state in the database.
     stores: RwLock<HashMap<String, Arc<ChainStore>>>,
     // We keep this information so we can create chain stores during startup
-    shards: Vec<(String, Shard)>,
+    shards: Vec<(String, Shard, BlockNumber)>,
+    // Per-chain cache_size settings
+    cache_sizes: HashMap<String, BlockNumber>,
     pools: HashMap<Shard, ConnectionPool>,
     sender: Arc<NotificationSender>,
     mirror: PrimaryMirror,
@@ -240,8 +242,8 @@ impl BlockStore {
     /// a chain uses the pool from `pools` for the given shard.
     pub async fn new(
         logger: Logger,
-        // (network, shard)
-        shards: Vec<(String, Shard)>,
+        // (network, shard, cache_size)
+        shards: Vec<(String, Shard, BlockNumber)>,
         // shard -> pool
         pools: HashMap<Shard, ConnectionPool>,
         sender: Arc<NotificationSender>,
@@ -257,6 +259,10 @@ impl BlockStore {
             .await?;
         let chain_head_cache = TimedCache::new(CHAIN_HEAD_CACHE_TTL);
         let chains = shards.clone();
+        let cache_sizes: HashMap<String, BlockNumber> = shards
+            .iter()
+            .map(|(name, _, cache_size)| (name.clone(), *cache_size))
+            .collect();
 
         let inner = Arc::new(Inner {
             logger,
@@ -267,11 +273,12 @@ impl BlockStore {
             mirror,
             chain_head_cache,
             chain_store_metrics,
+            cache_sizes,
         });
         let block_store = Self { inner };
 
         // For each configured chain, add a chain store
-        for (chain_name, shard) in chains {
+        for (chain_name, shard, _cache_size) in chains {
             if let Some(chain) = existing_chains
                 .iter()
                 .find(|chain| chain.name == chain_name)
@@ -363,6 +370,7 @@ impl BlockStore {
         );
         let ident = chain.network_identifier()?;
         let logger = self.logger.new(o!("network" => chain.name.clone()));
+        let cache_size = self.cache_sizes.get(&chain.name).copied().unwrap_or(500);
         let store = ChainStore::new(
             logger,
             chain.name.clone(),
@@ -371,6 +379,7 @@ impl BlockStore {
             pool,
             ENV_VARS.store.recent_blocks_cache_capacity,
             self.chain_store_metrics.clone(),
+            cache_size,
         );
         if create {
             store.create(&ident).await?;
@@ -565,7 +574,7 @@ impl BlockStore {
         let shard = self
             .shards
             .iter()
-            .find_map(|(chain_id, shard)| {
+            .find_map(|(chain_id, shard, _cache_size)| {
                 if chain_id.as_str().eq(network) {
                     Some(shard)
                 } else {

store/postgres/src/chain_store.rs

@@ -2194,6 +2194,9 @@ pub struct ChainStore {
     chain_head_ptr_cache: ChainHeadPtrCache,
     /// Herd cache to prevent thundering herd on chain_head_ptr() lookups
     chain_head_ptr_herd: HerdCache<Arc<Result<Option<BlockPtr>, StoreError>>>,
+    /// Number of blocks from chain head for which to keep block data cached.
+    /// Used with `GRAPH_STORE_IGNORE_BLOCK_CACHE` to simulate block data eviction.
+    cache_size: BlockNumber,
 }
 
 impl ChainStore {
@@ -2205,6 +2208,7 @@ impl ChainStore {
         pool: ConnectionPool,
         recent_blocks_cache_capacity: usize,
         metrics: Arc<ChainStoreMetrics>,
+        cache_size: BlockNumber,
     ) -> Self {
         let recent_blocks_cache =
             RecentBlocksCache::new(recent_blocks_cache_capacity, chain.clone(), metrics.clone());
@@ -2225,6 +2229,7 @@ impl ChainStore {
             ancestor_cache,
             chain_head_ptr_cache,
             chain_head_ptr_herd,
+            cache_size,
         }
     }