Merge pull request #1284 from movementlabsxyz/indexer-v2-4

larryl3u · web-flow · commit cf02a946a3db · 2025-08-18T09:05:38.000-07:00
[feat] Enable internal indexer db for grpc stream. (4/n)
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -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" }
diff --git a/docker/compose/movement-full-node/docker-compose.fullnode.yml b/docker/compose/movement-full-node/docker-compose.fullnode.yml
@@ -7,6 +7,20 @@ services:
       - DOT_MOVEMENT_PATH=/.movement
       - MOVEMENT_TIMING=info
       - RUST_BACKTRACE=1
+      - MOVEMENT_METRICS_ADDR=0.0.0.0:9464
+      - APTOS_FORCE_ENABLE_TELEMETRY=1
+      - APTOS_METRICS_PORT=9464
+      - APTOS_METRICS_HOST=0.0.0.0
+      - APTOS_DISABLE_TELEMETRY_PUSH_METRICS=1
+      - PROMETHEUS_METRICS_ENABLED=1
+      - APTOS_ENABLE_CONSENSUS_METRICS=1
+      - APTOS_ENABLE_DB_METRICS=1
+      - APTOS_ENABLE_MEMPOOL_METRICS=1
+      - APTOS_ENABLE_NETWORK_METRICS=1
+      - APTOS_ENABLE_STORAGE_METRICS=1
+      - APTOS_ENABLE_VM_METRICS=1
+      # OTEL Configuration (optional - uncomment and set your endpoint)
+      # - MOVEMENT_OTEL_ENDPOINT=http://otel-collector:4317
     volumes:
       - ${DOT_MOVEMENT_PATH}:/.movement
     ports:
diff --git a/docker/compose/movement-indexer/docker-compose.local-development.indexer-v2.yml b/docker/compose/movement-indexer/docker-compose.local-development.indexer-v2.yml
@@ -89,4 +89,4 @@ services:
 
 volumes:
   dot-movement:
-  db_data:
+  db_data:
diff --git a/networks/movement/indexer-v2/Cargo.toml b/networks/movement/indexer-v2/Cargo.toml
@@ -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]
diff --git a/networks/movement/indexer-v2/src/main.rs b/networks/movement/indexer-v2/src/main.rs
@@ -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)
 }
diff --git a/networks/movement/movement-full-node/src/main.rs b/networks/movement/movement-full-node/src/main.rs
@@ -2,6 +2,7 @@
 use clap::*;
 use movement_full_node::MovementFullNode;
 use tracing_subscriber::EnvFilter;
+
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
 	// Initialize default tracing
@@ -10,6 +11,12 @@ async fn main() -> Result<(), anyhow::Error> {
 			EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
 		)
 		.init();
+
+	// Initialize telemetry if MOVEMENT_METRICS_ADDR is set
+	if std::env::var("MOVEMENT_METRICS_ADDR").is_ok() {
+		movement_tracing::ensure_telemetry_initialized();
+	}
+
 	let suzuka_util = MovementFullNode::parse();
 	let result = suzuka_util.execute().await;
 	result
diff --git a/protocol-units/execution/maptos/opt-executor/Cargo.toml b/protocol-units/execution/maptos/opt-executor/Cargo.toml
@@ -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 }
diff --git a/protocol-units/execution/maptos/opt-executor/src/bootstrap.rs b/protocol-units/execution/maptos/opt-executor/src/bootstrap.rs
@@ -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,
 	)?;
diff --git a/protocol-units/execution/maptos/opt-executor/src/executor/initialization.rs b/protocol-units/execution/maptos/opt-executor/src/executor/initialization.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 {
diff --git a/protocol-units/execution/maptos/opt-executor/src/indexer.rs b/protocol-units/execution/maptos/opt-executor/src/indexer.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),
+		})
 	}
 }
diff --git a/protocol-units/execution/maptos/util/src/config/chain.rs b/protocol-units/execution/maptos/util/src/config/chain.rs
diff --git a/protocol-units/execution/maptos/util/src/config/common.rs b/protocol-units/execution/maptos/util/src/config/common.rs
diff --git a/util/tracing/src/lib.rs b/util/tracing/src/lib.rs
