Merge pull request #1297 from movementlabsxyz/indexer-v2-2

[feat] Enable internal indexer db for grpc stream. (4/n)

Author: larryl3u

Cargo.lock

@@ -175,6 +175,7 @@ aptos-crypto = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1
 
 aptos-crypto-derive = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
 aptos-db = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
+aptos-db-indexer = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
 aptos-executor = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
 aptos-executor-test-helpers = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
 aptos-executor-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1d1cdbbd7fabb80dcb95ba5e23213faa072fab67" }
@@ -206,9 +207,8 @@ movement = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "1d1cd
 # Indexer
 processor = { git = "https://github.com/movementlabsxyz/aptos-indexer-processors", rev = "20be1190105908fd4fea4e78c330997658e9428e" }
 server-framework = { git = "https://github.com/movementlabsxyz/aptos-indexer-processors", rev = "20be1190105908fd4fea4e78c330997658e9428e" }
-# TODO: use forked version.
-processor-v2 = { git = "https://github.com/larryl3u/aptos-indexer-processors-v2", rev = "2e5dd83639886fe9e44af97726d79895a3b4f650", package = "processor" }
-aptos-indexer-processor-sdk = { git = "https://github.com/larryl3u/aptos-indexer-processor-sdk", rev = "bbcc42d5a03cffe861b89c080ef7de918f37b4d0" }
+processor-v2 = { git = "https://github.com/movementlabsxyz/aptos-indexer-processors-v2", rev = "3fa850e7ea7770fec7967ce5b9a857dabace6aa3", package = "processor" }
+aptos-indexer-processor-sdk = { git = "https://github.com/movementlabsxyz/aptos-indexer-processor-sdk", rev = "a596af7bfdabeb4fc94c19bbbccbd0574d7c15a1" }
 
 
 bcs = { git = "https://github.com/movementlabsxyz/bcs.git", rev = "bc16d2d39cabafaabd76173dd1b04b2aa170cf0c" }

Cargo.toml

@@ -89,4 +89,4 @@ services:
 
 volumes:
   dot-movement:
-  db_data:
+  db_data:

docker/compose/movement-indexer/docker-compose.local-development.indexer-v2.yml

@@ -34,7 +34,7 @@ processor-v2 = { workspace = true }
 aptos-indexer-processor-sdk = { workspace = true }
 movement-tracing = { workspace = true }
 movement-health = { workspace = true }
-
+godfig = { workspace = true }
 serde_yaml = "0.9.34"
 
 [lints]

networks/movement/indexer-v2/Cargo.toml

@@ -1,4 +1,5 @@
 use aptos_indexer_processor_sdk::server_framework::RunnableConfig;
+use godfig::{backend::config_file::ConfigFile, Godfig};
 use maptos_execution_util::config::Config;
 use movement_health::run_service;
 use movement_tracing::simple_metrics::start_metrics_server;
@@ -10,27 +11,37 @@ fn main() -> Result<(), anyhow::Error> {
 	init_logger();
 
 	let runtime = get_maptos_runtime();
-	let maptos_config = load_maptos_config()?;
-	let runnable_processor_config: IndexerProcessorConfig =
-		maptos_config.indexer_processor_v2.clone().into();
 
 	runtime.block_on(async move {
+		let maptos_config = load_maptos_config().await.expect("Failed to load maptos config");
+		let runnable_processor_config: IndexerProcessorConfig =
+			maptos_config.indexer_processor_v2.clone().into();
+
 		let metrics_handle = tokio::spawn(async move {
-			start_metrics_server(
+			let res = start_metrics_server(
 				maptos_config.indexer_processor_v2.metrics_config.listen_hostname,
 				maptos_config.indexer_processor_v2.metrics_config.listen_port,
 			)
-			.await
+			.await;
+			tracing::info!("Metrics server started: {:?}", res);
+			res
 		});
 		let health_handle = tokio::spawn(async move {
-			run_service(
+			let res = run_service(
 				maptos_config.indexer_processor_v2.health_config.hostname,
 				maptos_config.indexer_processor_v2.health_config.port,
 			)
-			.await
+			.await;
+			tracing::info!("Health server started: {:?}", res);
+			res
 		});
 
-		let processor_handle = tokio::spawn(async move { runnable_processor_config.run().await });
+		let processor_handle = tokio::spawn(async move {
+			let res = runnable_processor_config.run().await;
+			tracing::info!("Indexer processor started: {:?}", res);
+			res
+		});
+		tracing::info!("Indexer v2 stack started.");
 
 		tokio::select! {
 			metrics_handle = metrics_handle => {
@@ -43,6 +54,7 @@ fn main() -> Result<(), anyhow::Error> {
 				tracing::error!("Indexer processor exited abnormally: {:?}", processor_handle);
 			}
 		}
+		tracing::info!("Indexer v2 stack exited normally.");
 	});
 	panic!("Indexer v2 stack exited abnormally.");
 }
@@ -78,8 +90,17 @@ fn get_maptos_runtime() -> tokio::runtime::Runtime {
 	runtime
 }
 
-fn load_maptos_config() -> anyhow::Result<Config> {
+async fn load_maptos_config() -> anyhow::Result<Config> {
+	// get the config file
 	let dot_movement = dot_movement::DotMovement::try_from_env()?;
-	let maptos_config = dot_movement.try_get_config_from_json::<Config>()?;
+
+	// Load Maptos config
+	let maptos_config = {
+		let config_file = dot_movement.try_get_or_create_config_file().await?;
+		let godfig: Godfig<maptos_execution_util::config::Config, ConfigFile> =
+			Godfig::new(ConfigFile::new(config_file), vec!["maptos_config".to_string()]);
+		godfig.try_wait_for_ready().await
+	}?;
+
 	Ok(maptos_config)
 }

networks/movement/indexer-v2/src/main.rs

@@ -44,6 +44,7 @@ aptos-db = { workspace = true }
 aptos-api = { workspace = true }
 aptos-api-types = { workspace = true }
 aptos-types = { workspace = true }
+aptos-db-indexer = { workspace = true }
 aptos-storage-interface = { workspace = true }
 aptos-block-executor = { workspace = true }
 aptos-vm-types = { workspace = true }

protocol-units/execution/maptos/opt-executor/Cargo.toml

@@ -86,13 +86,14 @@ pub fn maybe_bootstrap_empty_db(
 	chain_id: ChainId,
 	public_key: &Ed25519PublicKey,
 	release: &impl Release,
+	enable_indexer_db: bool,
 ) -> Result<(DbReaderWriter, ValidatorSigner), anyhow::Error> {
 	let aptos_db = AptosDB::open(
 		StorageDirPaths::from_path(db_dir.clone()),
 		false,
 		config.storage.storage_pruner_config.clone(),
 		config.storage.rocksdb_configs.clone(),
-		false,
+		enable_indexer_db,
 		config.storage.buffered_state_target_items,
 		config.storage.max_num_nodes_per_lru_cache_shard,
 	)?;

protocol-units/execution/maptos/opt-executor/src/bootstrap.rs

@@ -82,30 +82,25 @@ impl Executor {
 		node_config.indexer.batch_size = Some(8);
 		node_config.indexer.gap_lookback_versions = Some(4);
 
-		node_config.indexer_grpc.enabled = true;
-
 		// indexer_grpc config
-		if maptos_config.chain.enable_table_info_service {
+		if maptos_config.chain.enable_indexer_grpc {
 			node_config.indexer_table_info.enabled = true;
-			node_config.storage.enable_indexer = true;
+			node_config.indexer_grpc.enabled = true;
+			node_config.indexer_grpc.processor_batch_size = 4;
+			node_config.indexer_grpc.processor_task_count = 4;
+			node_config.indexer_grpc.output_batch_size = 4;
+			node_config.indexer_grpc.address = (
+				maptos_config.indexer.maptos_indexer_grpc_listen_hostname.as_str(),
+				maptos_config.indexer.maptos_indexer_grpc_listen_port,
+			)
+				.to_socket_addrs()?
+				.next()
+				.context("failed to resolve the value of maptos_indexer_grpc_listen_hostname")?;
+			node_config.indexer_grpc.use_data_service_interface = true;
 		}
-		node_config.indexer_grpc.processor_batch_size = 4;
-		node_config.indexer_grpc.processor_task_count = 4;
-		node_config.indexer_grpc.output_batch_size = 4;
-		node_config.indexer_grpc.address = (
-			maptos_config.indexer.maptos_indexer_grpc_listen_hostname.as_str(),
-			maptos_config.indexer.maptos_indexer_grpc_listen_port,
-		)
-			.to_socket_addrs()?
-			.next()
-			.context("failed to resolve the value of maptos_indexer_grpc_listen_hostname")?;
-		node_config.indexer_grpc.use_data_service_interface = true;
 
-		// indexer table info config
-		node_config.indexer_table_info.enabled = true;
 		node_config.storage.dir = dot_movement.get_path().join("maptos-storage");
 		node_config.storage.set_data_dir(node_config.storage.dir.clone());
-
 		let known_release = aptos_framework_known_release::KnownRelease::try_new(
 			maptos_config.chain.known_framework_release_str.as_str(),
 		)?;
@@ -115,6 +110,7 @@ impl Executor {
 			maptos_config.chain.maptos_chain_id.clone(),
 			&public_key,
 			&known_release,
+			maptos_config.chain.enable_indexer_grpc,
 		)?;
 
 		Ok(Self {

protocol-units/execution/maptos/opt-executor/src/executor/initialization.rs

@@ -1,44 +1,54 @@
 use super::Context;
+use aptos_db_indexer::indexer_reader::IndexerReaders;
 use aptos_indexer_grpc_fullnode::runtime::bootstrap as bootstrap_indexer_grpc;
 use aptos_indexer_grpc_table_info::runtime::bootstrap as bootstrap_table_info;
-
-use tokio::runtime::Runtime;
+use aptos_types::indexer::indexer_db_reader::IndexerReader;
+use std::sync::Arc;
 
 /// Runtime handle for indexer services. This object should be kept alive
 /// while services are running.
 pub struct IndexerRuntime {
-	// We only keep the runtimes around to drop them
-	_table_info_runtime: Runtime,
-	_indexer_grpc: Runtime,
-	//	_indexer_stream: Runtime,
+	_table_info_runtime: Option<tokio::runtime::Runtime>,
+	_indexer_grpc: Option<tokio::runtime::Runtime>,
 }
 
 impl Context {
 	// https://github.com/movementlabsxyz/aptos-core/blob/ea91067b81f9673547417bff9c70d5a2fe1b0e7b/aptos-node/src/services.rs#L40
 	pub fn run_indexer_grpc_service(&self) -> Result<IndexerRuntime, anyhow::Error> {
+		// If indexer grpc is not enabled, return empty runtime.
+		if !self.maptos_config.chain.enable_indexer_grpc {
+			return Ok(IndexerRuntime { _table_info_runtime: None, _indexer_grpc: None });
+		}
 		tracing::info!("Starting indexer gRPC service");
-
 		// bootstrap table info
-		let (_table_info_runtime, _async_indexer) = bootstrap_table_info(
+		let (table_info_runtime, async_indexer) = bootstrap_table_info(
 			&self.node_config,
 			self.maptos_config.chain.maptos_chain_id.clone(),
 			self.db.clone(),
 			self.mempool_client_sender.clone(),
 		)
 		.ok_or(anyhow::anyhow!("Failed to bootstrap table info runtime"))?;
+		// Indexer readers allow: 1. read from table info, 2 read from shard db. Shard db is not used thus None.
+		let indexer_readers = IndexerReaders::new(Some(async_indexer), None);
+		let indexer_reader: Option<Arc<dyn IndexerReader>> = indexer_readers.map(|readers| {
+			let trait_object: Arc<dyn IndexerReader> = Arc::new(readers);
+			trait_object
+		});
 
 		// Bootstrap indexer grpc, aka, transaction stream service.
 		// TSS requires the table info service and internal indexer db.
-		// TODO: add mempool client receiver to the bootstrap function.
-		let _indexer_grpc = bootstrap_indexer_grpc(
+		let indexer_grpc = bootstrap_indexer_grpc(
 			&self.node_config,
 			self.maptos_config.chain.maptos_chain_id.clone(),
 			self.db.reader.clone(),
 			self.mempool_client_sender.clone(),
-			None,
+			indexer_reader,
 		)
 		.ok_or(anyhow::anyhow!("Failed to bootstrap indexer grpc runtime"))?;
 
-		Ok(IndexerRuntime { _table_info_runtime, _indexer_grpc }) //, _indexer_stream
+		Ok(IndexerRuntime {
+			_table_info_runtime: Some(table_info_runtime),
+			_indexer_grpc: Some(indexer_grpc),
+		})
 	}
 }

protocol-units/execution/maptos/opt-executor/src/indexer.rs

@@ -1,5 +1,5 @@
 use super::common::{
-	default_enable_pruning, default_enable_table_info_service, default_genesis_block_hash_hex,
+	default_enable_indexer_grpc, default_enable_pruning, default_genesis_block_hash_hex,
 	default_genesis_timestamp_microseconds, default_maptos_chain_id,
 	default_maptos_epoch_snapshot_prune_window, default_maptos_ledger_prune_window,
 	default_maptos_private_key_signer_identifier, default_maptos_read_only,
@@ -82,9 +82,9 @@ pub struct Config {
 	#[serde(default = "default_dont_increase_epoch_until_version")]
 	pub dont_increase_epoch_until_version: u64,
 
-	/// Enable the table info service for indexer.
-	#[serde(default = "default_enable_table_info_service")]
-	pub enable_table_info_service: bool,
+	/// Enable the indexer grpc service.
+	#[serde(default = "default_enable_indexer_grpc")]
+	pub enable_indexer_grpc: bool,
 }
 
 impl Default for Config {
@@ -104,7 +104,7 @@ impl Default for Config {
 			maptos_db_path: None,
 			known_framework_release_str: default_known_framework_release_str(),
 			dont_increase_epoch_until_version: default_dont_increase_epoch_until_version(),
-			enable_table_info_service: default_enable_table_info_service(),
+			enable_indexer_grpc: default_enable_indexer_grpc(),
 		}
 	}
 }