feat(storage): implement its own GC method

chain/chain/Cargo.toml

@@ -18,6 +18,7 @@ num-rational = "0.2.4"
 tracing = "0.1.13"
 thiserror = "1.0"
 strum = "0.18"
+strum_macros = "0.18"
 
 borsh = "0.6.2"
 

chain/chain/src/chain.rs

@@ -952,13 +952,17 @@ impl Chain {
         chain_store_update.commit()?;
 
         // clear all trie data
-        let mut store_update = StoreUpdate::new_with_tries(self.runtime_adapter.get_tries());
-        let stored_state = self.store().store().iter_prefix(ColState, &[]);
-        for (key, _) in stored_state {
+        let keys: Vec<Vec<u8>> =
+            self.store().store().iter_prefix(ColState, &[]).map(|kv| kv.0.into()).collect();
+        let tries = self.runtime_adapter.get_tries();
+        let mut chain_store_update = self.mut_store().store_update();
+        let mut store_update = StoreUpdate::new_with_tries(tries);
+        for key in keys.iter() {
             store_update.delete(ColState, key.as_ref());
+            chain_store_update.inc_gc_col_state();
         }
-        let mut chain_store_update = self.mut_store().store_update();
         chain_store_update.merge(store_update);
+
         // The reason to reset tail here is not to allow Tail be greater than Head
         chain_store_update.reset_tail();
         chain_store_update.commit()?;
@@ -1793,12 +1797,9 @@ impl Chain {
         sync_hash: CryptoHash,
         num_parts: u64,
     ) -> Result<(), Error> {
-        let mut store_update = self.store.owned_store().store_update();
-        for part_id in 0..num_parts {
-            let key = StatePartKey(sync_hash, shard_id, part_id).try_to_vec()?;
-            store_update.delete(ColStateParts, &key);
-        }
-        Ok(store_update.commit()?)
+        let mut chain_store_update = self.mut_store().store_update();
+        chain_store_update.gc_col_state_parts(sync_hash, shard_id, num_parts)?;
+        Ok(chain_store_update.commit()?)
     }
 
     /// Apply transactions in chunks for the next epoch in blocks that were blocked on the state sync

core/primitives/src/version.rs

@@ -11,7 +11,7 @@ pub struct Version {
 pub type DbVersion = u32;
 
 /// Current version of the database.
-pub const DB_VERSION: DbVersion = 1;
+pub const DB_VERSION: DbVersion = 2;
 
 /// Protocol version type.
 pub type ProtocolVersion = u32;

core/store/src/db.rs

@@ -3,6 +3,8 @@ use std::collections::HashMap;
 use std::io;
 use std::sync::RwLock;
 
+use borsh::{BorshDeserialize, BorshSerialize};
+
 use rocksdb::{
     BlockBasedOptions, ColumnFamily, ColumnFamilyDescriptor, Direction, IteratorMode, Options,
     ReadOptions, WriteBatch, DB,
@@ -34,7 +36,7 @@ impl Into<io::Error> for DBError {
     }
 }
 
-#[derive(PartialEq, Debug, Copy, Clone, EnumIter)]
+#[derive(PartialEq, Debug, Copy, Clone, EnumIter, BorshDeserialize, BorshSerialize)]
 pub enum DBCol {
     /// Column to indicate which version of database this is.
     ColDbVersion = 0,
@@ -91,10 +93,12 @@ pub enum DBCol {
     ColChunkHashesByHeight = 39,
     /// Block ordinals.
     ColBlockOrdinal = 40,
+    /// GC Count for each column
+    ColGCCount = 41,
 }
 
 // Do not move this line from enum DBCol
-const NUM_COLS: usize = 41;
+pub const NUM_COLS: usize = 42;
 
 impl std::fmt::Display for DBCol {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
@@ -140,6 +144,7 @@ impl std::fmt::Display for DBCol {
             Self::ColBlockMerkleTree => "block merkle tree",
             Self::ColChunkHashesByHeight => "chunk hashes indexed by height_created",
             Self::ColBlockOrdinal => "block ordinal",
+            Self::ColGCCount => "gc count",
         };
         write!(formatter, "{}", desc)
     }

core/store/src/lib.rs

@@ -11,7 +11,7 @@ use cached::{Cached, SizedCache};
 pub use db::DBCol::{self, *};
 pub use db::{
     CHUNK_TAIL_KEY, HEADER_HEAD_KEY, HEAD_KEY, LARGEST_TARGET_HEIGHT_KEY, LATEST_KNOWN_KEY,
-    SYNC_HEAD_KEY, TAIL_KEY,
+    NUM_COLS, SYNC_HEAD_KEY, TAIL_KEY,
 };
 use near_crypto::PublicKey;
 use near_primitives::account::{AccessKey, Account};

neard/src/lib.rs

@@ -3,7 +3,7 @@ use std::path::Path;
 use std::sync::Arc;
 
 use actix::{Actor, Addr};
-use log::info;
+use log::{error, info};
 use tracing::trace;
 
 use near_chain::ChainGenesis;
@@ -53,8 +53,23 @@ pub fn get_default_home() -> String {
 
 /// Function checks current version of the database and applies migrations to the database.
 pub fn apply_store_migrations(path: &String) {
-    let _db_version = get_store_version(path);
+    let db_version = get_store_version(path);
+    if db_version > near_primitives::version::DB_VERSION {
+        error!(target: "near", "DB version {} is created by a newer version of neard, please update neard or delete data", db_version);
+        std::process::exit(1);
+    }
+    if db_version == near_primitives::version::DB_VERSION {
+        return;
+    }
     // Add migrations here based on `db_version`.
+    if db_version == 1 {
+        // version 1 => 2: add gc column
+        // Does not need to do anything since open db with option `create_missing_column_families`
+        // Nevertheless need to bump db version, because db_version 1 binary can't open db_version 2 db
+        info!(target: "near", "Migrate DB from version 1 to 2");
+        let store = create_store(&path);
+        set_store_version(&store);
+    }
 }
 
 pub fn init_and_migrate_store(home_dir: &Path) -> Arc<Store> {