[storage] split out aptos-indexer crate

Cargo.toml

@@ -105,6 +105,7 @@ members = [
     "storage/aptosdb",
     "storage/backup/backup-cli",
     "storage/backup/backup-service",
+    "storage/indexer",
     "storage/jellyfish-merkle",
     "storage/schemadb",
     "storage/scratchpad",

api/src/tests/test_context.rs

@@ -10,7 +10,7 @@ use aptos_api_types::{
     mime_types, HexEncodedBytes, TransactionOnChainData, X_APTOS_CHAIN_ID,
     X_APTOS_LEDGER_TIMESTAMP, X_APTOS_LEDGER_VERSION,
 };
-use aptos_config::config::NodeConfig;
+use aptos_config::config::{NodeConfig, RocksdbConfigs, NO_OP_STORAGE_PRUNER_CONFIG};
 use aptos_crypto::{hash::HashValue, SigningKey};
 use aptos_mempool::mocks::MockSharedMempool;
 use aptos_sdk::{
@@ -67,7 +67,16 @@ pub fn new_test_context(test_name: &'static str) -> TestContext {
     let (validator_identity, _, _) = validators[0].get_key_objects(None).unwrap();
     let validator_owner = validator_identity.account_address.unwrap();
 
-    let (db, db_rw) = DbReaderWriter::wrap(AptosDB::new_for_test(&tmp_dir));
+    let (db, db_rw) = DbReaderWriter::wrap(
+        AptosDB::open(
+            &tmp_dir,
+            false,                       /* readonly */
+            NO_OP_STORAGE_PRUNER_CONFIG, /* pruner */
+            RocksdbConfigs::default(),
+            true, /* indexer */
+        )
+        .unwrap(),
+    );
     let ret =
         db_bootstrapper::maybe_bootstrap::<AptosVM>(&db_rw, &genesis, genesis_waypoint).unwrap();
     assert!(ret);

aptos-move/aptos-validator-interface/src/storage_interface.rs

@@ -25,6 +25,7 @@ impl DBDebuggerInterface {
             true,
             NO_OP_STORAGE_PRUNER_CONFIG,
             RocksdbConfigs::default(),
+            false,
         )?)))
     }
 }

aptos-node/src/lib.rs

@@ -476,6 +476,7 @@ pub fn setup_environment(node_config: NodeConfig) -> anyhow::Result<AptosHandle>
             false, /* readonly */
             node_config.storage.storage_pruner_config,
             node_config.storage.rocksdb_configs,
+            node_config.storage.enable_indexer,
         )
         .map_err(|err| anyhow!("DB failed to open {}", err))?,
     );

config/src/config/storage_config.rs

@@ -77,6 +77,10 @@ pub struct StorageConfig {
     pub timeout_ms: u64,
     /// Rocksdb-specific configurations
     pub rocksdb_configs: RocksdbConfigs,
+    /// Try to enable the internal indexer. The indexer expects to have seen all transactions
+    /// since genesis. To recover operation after data loss, or to bootstrap a node in fast sync
+    /// mode, the indexer db needs to be copied in from another node.
+    pub enable_indexer: bool,
 }
 
 pub const NO_OP_STORAGE_PRUNER_CONFIG: StoragePrunerConfig = StoragePrunerConfig {
@@ -146,6 +150,7 @@ impl Default for StorageConfig {
             // Default read/write/connection timeout, in milliseconds
             timeout_ms: 30_000,
             rocksdb_configs: RocksdbConfigs::default(),
+            enable_indexer: false,
         }
     }
 }

crates/aptos-genesis/src/lib.rs

@@ -125,6 +125,7 @@ impl GenesisInfo {
             false,
             NO_OP_STORAGE_PRUNER_CONFIG,
             RocksdbConfigs::default(),
+            false,
         )?;
         let db_rw = DbReaderWriter::new(aptosdb);
         executor::db_bootstrapper::generate_waypoint::<AptosVM>(&db_rw, genesis)

execution/db-bootstrapper/src/bin/db-bootstrapper.rs

@@ -53,6 +53,7 @@ fn main() -> Result<()> {
             false,
             NO_OP_STORAGE_PRUNER_CONFIG, /* pruner */
             RocksdbConfigs::default(),
+            false, /* indexer */
         )
     } else {
         // When not committing, we open the DB as secondary so the tool is usable along side a

execution/executor-benchmark/src/db_generator.rs

@@ -58,6 +58,7 @@ fn bootstrap_with_genesis(db_dir: impl AsRef<Path>) {
             false, /* readonly */
             NO_OP_STORAGE_PRUNER_CONFIG,
             rocksdb_configs,
+            false, /* indexer */
         )
         .expect("DB should open."),
     );

execution/executor-benchmark/src/lib.rs

@@ -35,6 +35,7 @@ pub fn init_db_and_executor(config: &NodeConfig) -> (DbReaderWriter, BlockExecut
             false, /* readonly */
             config.storage.storage_pruner_config,
             RocksdbConfigs::default(),
+            false,
         )
         .expect("DB should open."),
     );
@@ -56,6 +57,7 @@ fn create_checkpoint(source_dir: impl AsRef<Path>, checkpoint_dir: impl AsRef<Pa
         true,                        /* readonly */
         NO_OP_STORAGE_PRUNER_CONFIG, /* pruner */
         RocksdbConfigs::default(),
+        false,
     )
     .expect("db open failure.")
     .create_checkpoint(checkpoint_dir.as_ref())

state-sync/state-sync-v2/state-sync-multiplexer/src/lib.rs

@@ -201,6 +201,7 @@ mod tests {
             false,
             NO_OP_STORAGE_PRUNER_CONFIG,
             RocksdbConfigs::default(),
+            false,
         )
         .unwrap();
         let (_, db_rw) = DbReaderWriter::wrap(db);

storage/aptosdb/Cargo.toml

@@ -36,6 +36,7 @@ aptos-state-view = { path = "../state-view" }
 aptos-temppath = { path = "../../crates/aptos-temppath", optional = true }
 aptos-types = { path = "../../types" }
 aptos-vm = { path = "../../aptos-move/aptos-vm" }
+aptosdb-indexer = { path = "../indexer" }
 
 executor-types = { path = "../../execution/executor-types" }
 move-deps = { path = "../../aptos-move/move-deps", features = ["address32"] }

storage/aptosdb/src/db_options.rs

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::schema::*;
-use aptos_config::config::RocksdbConfig;
 use aptos_types::transaction::Version;
 use schemadb::{
     ColumnFamilyDescriptor, ColumnFamilyName, DBCompressionType, Options, SliceTransform,
@@ -40,27 +39,6 @@ pub(super) fn state_merkle_db_column_families() -> Vec<ColumnFamilyName> {
     ]
 }
 
-pub(super) fn index_db_column_families() -> Vec<ColumnFamilyName> {
-    vec![
-        /* empty cf */ DEFAULT_COLUMN_FAMILY_NAME,
-        INDEXER_METADATA_CF_NAME,
-        TABLE_INFO_CF_NAME,
-    ]
-}
-
-pub(super) fn gen_rocksdb_options(config: &RocksdbConfig, readonly: bool) -> Options {
-    let mut db_opts = Options::default();
-    db_opts.set_max_open_files(config.max_open_files);
-    db_opts.set_max_total_wal_size(config.max_total_wal_size);
-    db_opts.set_max_background_jobs(config.max_background_jobs);
-    if !readonly {
-        db_opts.create_if_missing(true);
-        db_opts.create_missing_column_families(true);
-    }
-
-    db_opts
-}
-
 pub(super) fn gen_ledger_cfds() -> Vec<ColumnFamilyDescriptor> {
     let cfs = ledger_db_column_families();
     let mut cfds = Vec::with_capacity(cfs.len());
@@ -90,17 +68,6 @@ pub(super) fn gen_state_merkle_cfds() -> Vec<ColumnFamilyDescriptor> {
     cfds
 }
 
-pub(super) fn gen_index_db_cfds() -> Vec<ColumnFamilyDescriptor> {
-    let cfs = index_db_column_families();
-    let mut cfds = Vec::with_capacity(cfs.len());
-    for cf_name in cfs {
-        let mut cf_opts = Options::default();
-        cf_opts.set_compression_type(DBCompressionType::Lz4);
-        cfds.push(ColumnFamilyDescriptor::new((*cf_name).to_string(), cf_opts));
-    }
-    cfds
-}
-
 fn state_key_extractor(state_value_raw_key: &[u8]) -> &[u8] {
     &state_value_raw_key[..(state_value_raw_key.len() - VERSION_SIZE)]
 }