Migrate FA processor to SDK

rust/Cargo.toml

@@ -30,8 +30,8 @@ testing-transactions = { path = "testing-transactions" }
 
 ahash = { version = "0.8.7", features = ["serde"] }
 anyhow = "1.0.86"
-aptos-indexer-processor-sdk = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "9ecd252ccff53023664562001dd04c2886488c0d" }
-aptos-indexer-processor-sdk-server-framework = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "9ecd252ccff53023664562001dd04c2886488c0d" }
+aptos-indexer-processor-sdk = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "95d76e07dd66a20a0e50a9e6c559885bff7ab52b" }
+aptos-indexer-processor-sdk-server-framework = { git = "https://github.com/aptos-labs/aptos-indexer-processor-sdk.git", rev = "95d76e07dd66a20a0e50a9e6c559885bff7ab52b" }
 aptos-protos = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "5c48aee129b5a141be2792ffa3d9bd0a1a61c9cb" }
 aptos-system-utils = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "202bdccff2b2d333a385ae86a4fcf23e89da9f62" }
 aptos-indexer-test-transactions = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "7246f0536d599789d7dd5e9fb776554abbe11eac" }

rust/processor/src/processors/fungible_asset_processor.rs

@@ -169,7 +169,7 @@ async fn insert_to_db(
     Ok(())
 }
 
-fn insert_fungible_asset_activities_query(
+pub fn insert_fungible_asset_activities_query(
     items_to_insert: Vec<FungibleAssetActivity>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -186,7 +186,7 @@ fn insert_fungible_asset_activities_query(
     )
 }
 
-fn insert_fungible_asset_metadata_query(
+pub fn insert_fungible_asset_metadata_query(
     items_to_insert: Vec<FungibleAssetMetadataModel>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -222,7 +222,7 @@ fn insert_fungible_asset_metadata_query(
     )
 }
 
-fn insert_fungible_asset_balances_query(
+pub fn insert_fungible_asset_balances_query(
     items_to_insert: Vec<FungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -239,7 +239,7 @@ fn insert_fungible_asset_balances_query(
     )
 }
 
-fn insert_current_fungible_asset_balances_query(
+pub fn insert_current_fungible_asset_balances_query(
     items_to_insert: Vec<CurrentFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -269,7 +269,7 @@ fn insert_current_fungible_asset_balances_query(
     )
 }
 
-fn insert_current_unified_fungible_asset_balances_v1_query(
+pub fn insert_current_unified_fungible_asset_balances_v1_query(
     items_to_insert: Vec<CurrentUnifiedFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -298,7 +298,7 @@ fn insert_current_unified_fungible_asset_balances_v1_query(
     )
 }
 
-fn insert_current_unified_fungible_asset_balances_v2_query(
+pub fn insert_current_unified_fungible_asset_balances_v2_query(
     items_to_insert: Vec<CurrentUnifiedFungibleAssetBalance>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -328,7 +328,7 @@ fn insert_current_unified_fungible_asset_balances_v2_query(
     )
 }
 
-fn insert_coin_supply_query(
+pub fn insert_coin_supply_query(
     items_to_insert: Vec<CoinSupply>,
 ) -> (
     impl QueryFragment<Pg> + diesel::query_builder::QueryId + Send,
@@ -451,7 +451,7 @@ impl ProcessorTrait for FungibleAssetProcessor {
 }
 
 /// V2 coin is called fungible assets and this flow includes all data from V1 in coin_processor
-async fn parse_v2_coin(
+pub async fn parse_v2_coin(
     transactions: &[Transaction],
 ) -> (
     Vec<FungibleAssetActivity>,

rust/processor/src/worker.rs

@@ -114,6 +114,18 @@ bitflags! {
     }
 }
 
+impl TableFlags {
+    pub fn from_set(set: &HashSet<String>) -> Self {
+        let mut flags = TableFlags::empty();
+        for table in set {
+            if let Some(flag) = TableFlags::from_name(table) {
+                flags |= flag;
+            }
+        }
+        flags
+    }
+}
+
 pub struct Worker {
     pub db_pool: ArcDbPool,
     pub processor_config: ProcessorConfig,

rust/sdk-processor/src/config/indexer_processor_config.rs

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::{db_config::DbConfig, processor_config::ProcessorConfig};
-use crate::processors::events_processor::EventsProcessor;
+use crate::processors::{
+    events_processor::EventsProcessor, fungible_asset_processor::FungibleAssetProcessor,
+};
 use anyhow::Result;
 use aptos_indexer_processor_sdk::aptos_indexer_transaction_stream::TransactionStreamConfig;
 use aptos_indexer_processor_sdk_server_framework::RunnableConfig;
@@ -24,6 +26,10 @@ impl RunnableConfig for IndexerProcessorConfig {
                 let events_processor = EventsProcessor::new(self.clone()).await?;
                 events_processor.run_processor().await
             },
+            ProcessorConfig::FungibleAssetProcessor(_) => {
+                let fungible_asset_processor = FungibleAssetProcessor::new(self.clone()).await?;
+                fungible_asset_processor.run_processor().await
+            },
         }
     }
 

rust/sdk-processor/src/config/processor_config.rs

@@ -1,5 +1,6 @@
-use crate::processors::events_processor::EventsProcessorConfig;
+use ahash::AHashMap;
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 
 /// This enum captures the configs for all the different processors that are defined.
 ///
@@ -34,7 +35,8 @@ use serde::{Deserialize, Serialize};
     strum(serialize_all = "snake_case")
 )]
 pub enum ProcessorConfig {
-    EventsProcessor(EventsProcessorConfig),
+    EventsProcessor(DefaultProcessorConfig),
+    FungibleAssetProcessor(DefaultProcessorConfig),
 }
 
 impl ProcessorConfig {
@@ -44,33 +46,23 @@ impl ProcessorConfig {
         self.into()
     }
 }
-#[derive(Debug)]
-// To ensure that the variants of ProcessorConfig and Processor line up, in the testing
-// build path we derive EnumDiscriminants on this enum as well and make sure the two
-// sets of variants match up in `test_processor_names_complete`.
-#[cfg_attr(
-    test,
-    derive(strum::EnumDiscriminants),
-    strum_discriminants(
-        derive(strum::EnumVariantNames),
-        name(ProcessorDiscriminants),
-        strum(serialize_all = "snake_case")
-    )
-)]
-pub enum Processor {
-    EventsProcessor,
-}
 
-#[cfg(test)]
-mod test {
-    use super::*;
-    use strum::VariantNames;
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct DefaultProcessorConfig {
+    // Number of rows to insert, per chunk, for each DB table. Default per table is ~32,768 (2**16/2)
+    #[serde(default = "AHashMap::new")]
+    pub per_table_chunk_sizes: AHashMap<String, usize>,
+    // Size of channel between steps
+    #[serde(default = "DefaultProcessorConfig::default_channel_size")]
+    pub channel_size: usize,
+    // String vector for deprecated tables to skip db writes
+    #[serde(default)]
+    pub deprecated_tables: HashSet<String>,
+}
 
-    /// This test exists to make sure that when a new processor is added, it is added
-    /// to both Processor and ProcessorConfig. To make sure this passes, make sure the
-    /// variants are in the same order (lexicographical) and the names match.
-    #[test]
-    fn test_processor_names_complete() {
-        assert_eq!(ProcessorName::VARIANTS, ProcessorDiscriminants::VARIANTS);
+impl DefaultProcessorConfig {
+    pub const fn default_channel_size() -> usize {
+        10
     }
 }

rust/sdk-processor/src/processors/events_processor.rs

@@ -13,34 +13,15 @@ use crate::{
         starting_version::get_starting_version,
     },
 };
-use ahash::AHashMap;
 use anyhow::Result;
 use aptos_indexer_processor_sdk::{
     aptos_indexer_transaction_stream::{TransactionStream, TransactionStreamConfig},
     builder::ProcessorBuilder,
     common_steps::TransactionStreamStep,
     traits::IntoRunnableStep,
 };
-use serde::{Deserialize, Serialize};
 use tracing::{debug, info};
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
-#[serde(deny_unknown_fields)]
-pub struct EventsProcessorConfig {
-    // Number of rows to insert, per chunk, for each DB table. Default per table is ~32,768 (2**16/2)
-    #[serde(default = "AHashMap::new")]
-    pub per_table_chunk_sizes: AHashMap<String, usize>,
-    // Size of channel between steps
-    #[serde(default = "EventsProcessorConfig::default_channel_size")]
-    pub channel_size: usize,
-}
-
-impl EventsProcessorConfig {
-    pub const fn default_channel_size() -> usize {
-        10
-    }
-}
-
 pub struct EventsProcessor {
     pub config: IndexerProcessorConfig,
     pub db_pool: ArcDbPool,
@@ -73,7 +54,7 @@ impl EventsProcessor {
     pub async fn run_processor(self) -> Result<()> {
         let processor_name = self.config.processor_config.name();
 
-        // (Optional) Run migrations
+        // Run migrations
         match self.config.db_config {
             DbConfig::PostgresConfig(ref postgres_config) => {
                 run_migrations(
@@ -84,19 +65,26 @@ impl EventsProcessor {
             },
         }
 
-        // (Optional) Merge the starting version from config and the latest processed version from the DB
+        //  Merge the starting version from config and the latest processed version from the DB
         let starting_version = get_starting_version(&self.config, self.db_pool.clone()).await?;
 
-        // (Optional) Check and update the ledger chain id to ensure we're indexing the correct chain
+        // Check and update the ledger chain id to ensure we're indexing the correct chain
         let grpc_chain_id = TransactionStream::new(self.config.transaction_stream_config.clone())
             .await?
             .get_chain_id()
             .await?;
         check_or_update_chain_id(grpc_chain_id as i64, self.db_pool.clone()).await?;
 
-        let ProcessorConfig::EventsProcessor(events_processor_config) =
-            self.config.processor_config;
-        let channel_size = events_processor_config.channel_size;
+        let processor_config = match self.config.processor_config {
+            ProcessorConfig::EventsProcessor(processor_config) => processor_config,
+            _ => {
+                return Err(anyhow::anyhow!(
+                    "Invalid processor config for EventsProcessor: {:?}",
+                    self.config.processor_config
+                ))
+            },
+        };
+        let channel_size = processor_config.channel_size;
 
         // Define processor steps
         let transaction_stream = TransactionStreamStep::new(TransactionStreamConfig {
@@ -105,7 +93,7 @@ impl EventsProcessor {
         })
         .await?;
         let events_extractor = EventsExtractor {};
-        let events_storer = EventsStorer::new(self.db_pool.clone(), events_processor_config);
+        let events_storer = EventsStorer::new(self.db_pool.clone(), processor_config);
         let version_tracker = LatestVersionProcessedTracker::new(
             self.db_pool.clone(),
             starting_version,
@@ -125,12 +113,9 @@ impl EventsProcessor {
         loop {
             match buffer_receiver.recv().await {
                 Ok(txn_context) => {
-                    if txn_context.data.is_empty() {
-                        continue;
-                    }
                     debug!(
                         "Finished processing events from versions [{:?}, {:?}]",
-                        txn_context.start_version, txn_context.end_version,
+                        txn_context.metadata.start_version, txn_context.metadata.end_version,
                     );
                 },
                 Err(e) => {