store: introduce cold storage (#7871)

Add a `cold_store` Cargo feature which enables the option to configure
the node with cold storage.  At the moment, all this does is open the
cold database and doesn’t enable any other features.  The idea is that
this can now allow experimenting with code that needs access to the
cold storage.

chain/indexer/src/streamer/mod.rs

@@ -290,9 +290,10 @@ pub(crate) async fn start(
     blocks_sink: mpsc::Sender<StreamerMessage>,
 ) {
     info!(target: INDEXER, "Starting Streamer...");
-    let indexer_db_path = near_store::NodeStorage::opener(&indexer_config.home_dir, &store_config)
-        .path()
-        .join("indexer");
+    let indexer_db_path =
+        near_store::NodeStorage::opener(&indexer_config.home_dir, &store_config, None)
+            .path()
+            .join("indexer");
 
     // TODO: implement proper error handling
     let db = DB::open_default(indexer_db_path).unwrap();

core/store/Cargo.toml

@@ -57,6 +57,7 @@ no_cache = []
 single_thread_rocksdb = [] # Deactivate RocksDB IO background threads
 test_features = []
 protocol_feature_flat_state = []
+cold_store = []
 
 nightly_protocol = []
 nightly = [

core/store/benches/store_bench.rs

@@ -19,10 +19,15 @@ fn benchmark_write_then_read_successful(
     let tmp_dir = tempfile::tempdir().unwrap();
     // Use default StoreConfig rather than NodeStorage::test_opener so we’re using the
     // same configuration as in production.
-    let store = NodeStorage::opener(tmp_dir.path(), &Default::default())
-        .open()
-        .unwrap()
-        .get_store(Temperature::Hot);
+    let store = NodeStorage::opener(
+        tmp_dir.path(),
+        &Default::default(),
+        #[cfg(feature = "cold_store")]
+        None,
+    )
+    .open()
+    .unwrap()
+    .get_store(Temperature::Hot);
     let keys = generate_keys(num_keys, key_size);
     write_to_db(&store, &keys, max_value_size, col);
 

core/store/src/db.rs

@@ -2,12 +2,15 @@ use std::io;
 
 use crate::DBCol;
 
+#[cfg(feature = "cold_store")]
 mod colddb;
 pub mod refcount;
 pub(crate) mod rocksdb;
 mod slice;
 mod testdb;
 
+#[cfg(feature = "cold_store")]
+pub use self::colddb::ColdDB;
 pub use self::rocksdb::RocksDB;
 pub use self::slice::DBSlice;
 pub use self::testdb::TestDB;

core/store/src/db/colddb.rs

@@ -35,9 +35,15 @@ use crate::DBCol;
 /// Lastly, since no data is ever deleted from cold storage, trying to decrease
 /// reference of a value count or delete data is ignored and if debug assertions
 /// are enabled will cause a panic.
-struct ColdDatabase<D: Database = crate::db::RocksDB>(D);
+pub struct ColdDB<D: Database = crate::db::RocksDB>(D);
 
-impl<D: Database> ColdDatabase<D> {
+impl<D: Database> std::convert::From<D> for ColdDB<D> {
+    fn from(db: D) -> Self {
+        Self(db)
+    }
+}
+
+impl<D: Database> ColdDB<D> {
     /// Returns raw bytes from the underlying storage.
     ///
     /// Adjusts the key if necessary (see [`get_cold_key`]) and retrieves data
@@ -51,7 +57,7 @@ impl<D: Database> ColdDatabase<D> {
     }
 }
 
-impl<D: Database> super::Database for ColdDatabase<D> {
+impl<D: Database> super::Database for ColdDB<D> {
     fn get_raw_bytes(&self, col: DBCol, key: &[u8]) -> std::io::Result<Option<DBSlice<'_>>> {
         match self.get_impl(col, key) {
             Ok(Some(value)) if col.is_rc() => {
@@ -262,8 +268,8 @@ mod test {
     const VALUE: &[u8] = "FooBar".as_bytes();
 
     /// Constructs test in-memory database.
-    fn create_test_db() -> ColdDatabase<crate::db::TestDB> {
-        ColdDatabase(crate::db::testdb::TestDB::default())
+    fn create_test_db() -> ColdDB<crate::db::TestDB> {
+        ColdDB(crate::db::testdb::TestDB::default())
     }
 
     fn set(col: DBCol, key: &[u8]) -> DBOp {
@@ -338,8 +344,8 @@ mod test {
 
                 let name = if is_raw { "raw " } else { "cold" };
                 let value = db.get_raw_bytes(col, &key).unwrap();
-                // When fetching reference counted column ColdDatabase adds
-                // reference count to it.
+                // When fetching reference counted column ColdDB adds reference
+                // count to it.
                 let value = pretty_value(value.as_deref(), col.is_rc() && !is_raw);
                 result.push(format!("    [{name}] get_raw_bytes → {value}"));
 

core/store/src/db/rocksdb/snapshot.rs

@@ -190,7 +190,7 @@ fn test_snapshot_recovery() {
     {
         let mut config = opener.config().clone();
         config.path = Some(path);
-        let opener = crate::NodeStorage::opener(tmpdir.path(), &config);
+        let opener = crate::NodeStorage::opener(tmpdir.path(), &config, None);
         let store = opener.open().unwrap().get_store(crate::Temperature::Hot);
         assert_eq!(Some(&b"value"[..]), store.get(COL, KEY).unwrap().as_deref());
     }

core/store/src/lib.rs

@@ -57,8 +57,11 @@ pub use crate::opener::{StoreMigrator, StoreOpener, StoreOpenerError};
 /// In the future, certain parts of the code may need to access hot or cold
 /// storage.  Specifically, querying an old block will require reading it from
 /// the cold storage.
+#[derive(Clone, Copy)]
 pub enum Temperature {
     Hot,
+    #[cfg(feature = "cold_store")]
+    Cold,
 }
 
 /// Node’s storage holding chain and all other necessary data.
@@ -68,7 +71,11 @@ pub enum Temperature {
 /// to [`Store`] which will abstract access to only one of the temperatures of
 /// the storage.
 pub struct NodeStorage {
-    storage: Arc<dyn Database>,
+    hot_storage: Arc<dyn Database>,
+    #[cfg(feature = "cold_store")]
+    cold_storage: Option<Arc<crate::db::ColdDB>>,
+    #[cfg(not(feature = "cold_store"))]
+    cold_storage: Option<std::convert::Infallible>,
 }
 
 /// Node’s single storage source.
@@ -84,10 +91,22 @@ pub struct Store {
     storage: Arc<dyn Database>,
 }
 
+// Those are temporary.  While cold_store feature is stabilised, remove those
+// type aliases and just use the type directly.
+#[cfg(feature = "cold_store")]
+pub type ColdConfig<'a> = Option<&'a StoreConfig>;
+#[cfg(not(feature = "cold_store"))]
+pub type ColdConfig<'a> = Option<std::convert::Infallible>;
+
 impl NodeStorage {
-    /// Initialises a new opener with given home directory and store config.
-    pub fn opener<'a>(home_dir: &std::path::Path, config: &'a StoreConfig) -> StoreOpener<'a> {
-        StoreOpener::new(home_dir, config)
+    /// Initialises a new opener with given home directory and hot and cold
+    /// store config.
+    pub fn opener<'a>(
+        home_dir: &std::path::Path,
+        config: &'a StoreConfig,
+        cold_config: ColdConfig<'a>,
+    ) -> StoreOpener<'a> {
+        StoreOpener::new(home_dir, config, cold_config)
     }
 
     /// Initialises an opener for a new temporary test store.
@@ -101,7 +120,7 @@ impl NodeStorage {
     pub fn test_opener() -> (tempfile::TempDir, StoreOpener<'static>) {
         static CONFIG: Lazy<StoreConfig> = Lazy::new(StoreConfig::test_config);
         let dir = tempfile::tempdir().unwrap();
-        let opener = StoreOpener::new(dir.path(), &CONFIG);
+        let opener = StoreOpener::new(dir.path(), &CONFIG, None);
         (dir, opener)
     }
 
@@ -115,7 +134,22 @@ impl NodeStorage {
     /// possibly [`crate::test_utils::create_test_store`] (depending whether you
     /// need [`NodeStorage`] or [`Store`] object.
     pub fn new(storage: Arc<dyn Database>) -> Self {
-        Self { storage }
+        Self { hot_storage: storage, cold_storage: None }
+    }
+
+    /// Constructs new object backed by given database.
+    fn from_rocksdb(
+        hot_storage: crate::db::RocksDB,
+        #[cfg(feature = "cold_store")] cold_storage: Option<crate::db::RocksDB>,
+        #[cfg(not(feature = "cold_store"))] cold_storage: Option<std::convert::Infallible>,
+    ) -> Self {
+        Self {
+            hot_storage: Arc::new(hot_storage),
+            #[cfg(feature = "cold_store")]
+            cold_storage: cold_storage.map(|db| Arc::new(db.into())),
+            #[cfg(not(feature = "cold_store"))]
+            cold_storage: cold_storage.map(|_| unreachable!()),
+        }
     }
 
     /// Returns storage for given temperature.
@@ -132,7 +166,9 @@ impl NodeStorage {
     /// cold.
     pub fn get_store(&self, temp: Temperature) -> Store {
         match temp {
-            Temperature::Hot => Store { storage: self.storage.clone() },
+            Temperature::Hot => Store { storage: self.hot_storage.clone() },
+            #[cfg(feature = "cold_store")]
+            Temperature::Cold => Store { storage: self.cold_storage.as_ref().unwrap().clone() },
         }
     }
 
@@ -151,9 +187,11 @@ impl NodeStorage {
     /// well.  For example, garbage collection only ever touches hot storage but
     /// it should go through [`Store`] interface since data it manipulates
     /// (e.g. blocks) are live in both databases.
-    pub fn get_inner(&self, temp: Temperature) -> &Arc<dyn Database> {
+    pub fn _get_inner(&self, temp: Temperature) -> &Arc<dyn Database> {
         match temp {
-            Temperature::Hot => &self.storage,
+            Temperature::Hot => &self.hot_storage,
+            #[cfg(feature = "cold_store")]
+            Temperature::Cold => todo!(),
         }
     }
 
@@ -163,15 +201,27 @@ impl NodeStorage {
     /// `Arc::clone`.
     pub fn into_inner(self, temp: Temperature) -> Arc<dyn Database> {
         match temp {
-            Temperature::Hot => self.storage,
+            Temperature::Hot => self.hot_storage,
+            #[cfg(feature = "cold_store")]
+            Temperature::Cold => self.cold_storage.unwrap(),
         }
     }
 
+    /// Returns whether the storage has a cold database.
+    pub fn has_cold(&self) -> bool {
+        self.cold_storage.is_some()
+    }
+
     /// Reads database metadata and returns whether the storage is archival.
     pub fn is_archive(&self) -> io::Result<bool> {
-        Ok(match metadata::DbMetadata::read(self.storage.as_ref())?.kind.unwrap() {
+        if self.cold_storage.is_some() {
+            return Ok(true);
+        }
+        Ok(match metadata::DbMetadata::read(self.hot_storage.as_ref())?.kind.unwrap() {
             metadata::DbKind::RPC => false,
             metadata::DbKind::Archive => true,
+            #[cfg(feature = "cold_store")]
+            metadata::DbKind::Hot | metadata::DbKind::Cold => unreachable!(),
         })
     }
 }

core/store/src/metadata.rs

@@ -35,28 +35,53 @@ pub enum DbKind {
     /// The database is an archive database meaning that it is not garbage
     /// collected and stores all chain data.
     Archive,
+    #[cfg(feature = "cold_store")]
+    /// The database is Hot meaning that the node runs in archival mode with
+    /// a paired Cold database.
+    Hot,
+    #[cfg(feature = "cold_store")]
+    /// The database is Cold meaning that the node runs in archival mode with
+    /// a paired Hot database.
+    Cold,
 }
 
 pub(super) fn set_store_version(storage: &NodeStorage, version: DbVersion) -> std::io::Result<()> {
     set_store_metadata(storage, DbMetadata { version, kind: None })
 }
 
-pub(super) fn set_store_metadata(
+fn set_db_metadata(
     storage: &NodeStorage,
+    temp: Temperature,
     metadata: DbMetadata,
 ) -> std::io::Result<()> {
-    let version = metadata.version.to_string().into_bytes();
-    let kind = metadata.kind.map(|kind| <&str>::from(kind).as_bytes());
-    let mut store_update = storage.get_store(Temperature::Hot).store_update();
-    store_update.set(DBCol::DbVersion, VERSION_KEY, &version);
+    let mut store_update = storage.get_store(temp).store_update();
+    store_update.set(DBCol::DbVersion, VERSION_KEY, metadata.version.to_string().as_bytes());
     if metadata.version >= DB_VERSION_WITH_KIND {
-        if let Some(kind) = kind {
+        #[allow(unused_mut)]
+        let mut kind = metadata.kind;
+        #[cfg(feature = "cold_store")]
+        if matches!(temp, Temperature::Cold) || storage.has_cold() {
+            kind = Some(if matches!(temp, Temperature::Hot) { DbKind::Hot } else { DbKind::Cold });
+        }
+        if let Some(kind) = kind.map(|kind| <&str>::from(kind).as_bytes()) {
             store_update.set(DBCol::DbVersion, KIND_KEY, kind);
         }
     }
     store_update.commit()
 }
 
+pub(super) fn set_store_metadata(
+    storage: &NodeStorage,
+    metadata: DbMetadata,
+) -> std::io::Result<()> {
+    set_db_metadata(storage, Temperature::Hot, metadata)?;
+    #[cfg(feature = "cold_store")]
+    if storage.has_cold() {
+        set_db_metadata(storage, Temperature::Cold, metadata)?;
+    }
+    Ok(())
+}
+
 /// Metadata about a database.
 #[derive(Clone, Copy)]
 pub(super) struct DbMetadata {