node, store: Update db_version to version 3 at startup

Adds a startup operation to check if `db_version` is less than 3.

If not, performs full cleanup on all block cache tables (using TRUNCATE TABLE) and then set the `db_version` to 3.

Author: tilacog

node/src/store_builder.rs

@@ -166,11 +166,14 @@ impl StoreBuilder {
             )
             .expect("Creating the BlockStore works"),
         );
+        block_store
+            .update_db_version()
+            .expect("Updating `db_version` works");
 
         Arc::new(DieselStore::new(subgraph_store, block_store))
     }
 
-    /// Create a connection pool for the main database of hte primary shard
+    /// Create a connection pool for the main database of the primary shard
     /// without connecting to all the other configured databases
     pub fn main_pool(
         logger: &Logger,

store/postgres/src/block_store.rs

@@ -21,7 +21,7 @@ use graph::{
 
 use crate::{
     chain_head_listener::ChainHeadUpdateSender, connection_pool::ConnectionPool,
-    primary::Mirror as PrimaryMirror, ChainStore, NotificationSender, Shard,
+    primary::Mirror as PrimaryMirror, ChainStore, NotificationSender, Shard, PRIMARY_SHARD,
 };
 
 #[cfg(debug_assertions)]
@@ -462,6 +462,29 @@ impl BlockStore {
 
         Ok(())
     }
+
+    fn truncate_block_caches(&self) -> Result<(), StoreError> {
+        for (_chain, store) in &*self.stores.read().unwrap() {
+            store.truncate_block_cache()?
+        }
+        Ok(())
+    }
+
+    pub fn update_db_version(&self) -> Result<(), StoreError> {
+        use crate::primary::db_version as dbv;
+        use diesel::prelude::*;
+
+        let primary_pool = self.pools.get(&*PRIMARY_SHARD).unwrap();
+        let connection = primary_pool.get()?;
+        let version: i64 = dbv::table.select(dbv::version).get_result(&connection)?;
+        if version < 3 {
+            self.truncate_block_caches()?;
+            diesel::update(dbv::table)
+                .set(dbv::version.eq(3))
+                .execute(&connection)?;
+        };
+        Ok(())
+    }
 }
 
 impl BlockStoreTrait for BlockStore {

store/postgres/src/chain_store.rs

@@ -390,6 +390,15 @@ mod data {
             }
         }
 
+        pub(super) fn truncate_block_cache(&self, conn: &PgConnection) -> Result<(), StoreError> {
+            let table_name = match &self {
+                Storage::Shared => ETHEREUM_BLOCKS_TABLE_NAME,
+                Storage::Private(Schema { blocks, .. }) => &blocks.qname,
+            };
+            conn.batch_execute(&format!("truncate table {} restart identity", table_name))?;
+            Ok(())
+        }
+
         /// Insert a block. If the table already contains a block with the
         /// same hash, then overwrite that block since it may be adding
         /// transaction receipts. If `overwrite` is `true`, overwrite a
@@ -1264,6 +1273,12 @@ impl ChainStore {
         self.storage
             .set_chain(&conn, &self.chain, genesis_hash, chain);
     }
+
+    pub fn truncate_block_cache(&self) -> Result<(), StoreError> {
+        let conn = self.get_conn()?;
+        self.storage.truncate_block_cache(&conn)?;
+        Ok(())
+    }
 }
 
 #[async_trait]

store/postgres/src/lib.rs

@@ -66,7 +66,7 @@ pub use self::chain_store::ChainStore;
 pub use self::detail::DeploymentDetail;
 pub use self::jobs::register as register_jobs;
 pub use self::notification_listener::NotificationSender;
-pub use self::primary::UnusedDeployment;
+pub use self::primary::{db_version, UnusedDeployment};
 pub use self::store::Store;
 pub use self::store_events::SubscriptionManager;
 pub use self::subgraph_store::{unused, DeploymentPlacer, Shard, SubgraphStore, PRIMARY_SHARD};

store/postgres/src/primary.rs

@@ -171,6 +171,13 @@ table! {
     }
 }
 
+table! {
+    public.db_version(version) {
+        #[sql_name = "db_version"]
+        version -> BigInt,
+    }
+}
+
 allow_tables_to_appear_in_same_query!(
     subgraph,
     subgraph_version,