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

@@ -1,6 +1,7 @@
 use std::iter::FromIterator;
 use std::{collections::HashMap, sync::Arc};
 
+use graph::components::store::BLOCK_CACHE_SIZE;
 use graph::prelude::{o, MetricsRegistry, NodeId};
 use graph::slog::warn;
 use graph::url::Url;
@@ -24,7 +25,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 +67,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 +179,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(), BLOCK_CACHE_SIZE));
+                (name, shard, cache_size)
             })
             .collect();
 

store/postgres/src/block_store.rs

@@ -1,6 +1,6 @@
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use graph::parking_lot::RwLock;
+use graph::{components::store::BLOCK_CACHE_SIZE, parking_lot::RwLock};
 
 use anyhow::anyhow;
 use async_trait::async_trait;
@@ -217,8 +217,9 @@ pub struct Inner {
     /// known to the system at startup, either from configuration or from
     /// 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)>,
+    /// We keep this information so we can create chain stores during
+    /// startup. The triple is (network, shard, cache_size)
+    shards: Vec<(String, Shard, BlockNumber)>,
     pools: HashMap<Shard, ConnectionPool>,
     sender: Arc<NotificationSender>,
     mirror: PrimaryMirror,
@@ -240,8 +241,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>,
@@ -271,7 +272,7 @@ impl BlockStore {
         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 +364,17 @@ impl BlockStore {
         );
         let ident = chain.network_identifier()?;
         let logger = self.logger.new(o!("network" => chain.name.clone()));
+        let cache_size = self
+            .shards
+            .iter()
+            .find_map(|(network, _, chain_size)| {
+                if network == &chain.name {
+                    Some(*chain_size)
+                } else {
+                    None
+                }
+            })
+            .unwrap_or(BLOCK_CACHE_SIZE);
         let store = ChainStore::new(
             logger,
             chain.name.clone(),
@@ -371,6 +383,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 +578,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

@@ -2347,6 +2347,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 {
@@ -2358,6 +2361,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());
@@ -2378,6 +2382,7 @@ impl ChainStore {
             ancestor_cache,
             chain_head_ptr_cache,
             chain_head_ptr_herd,
+            cache_size,
         }
     }