[mtg-836] Separate fungible token accounts from non-fungible ones

entities/src/models.rs

@@ -73,6 +73,17 @@ pub struct AssetIndex {
     pub fungible_asset_balance: Option<u64>,
 }
 
+/// FungibleAssetIndex is the struct that is stored in the postgres, and is used to query the fungible asset pubkeys.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
+pub struct FungibleAssetIndex {
+    pub pubkey: Pubkey,
+    pub owner: Option<Pubkey>,
+    pub specification_asset_class: SpecificationAssetClass,
+    pub slot_updated: i64,
+    pub fungible_asset_mint: Option<Pubkey>,
+    pub fungible_asset_balance: Option<u64>,
+}
+
 /// FungibleToken is associated token account
 /// owned by some user
 /// key - token account's pubkey

migrations/1_init.sql

@@ -123,4 +123,4 @@ CREATE TABLE last_synced_key (
 );
 
 -- Insert an initial row (assuming there's no last_synced_key initially)
-INSERT INTO last_synced_key (last_synced_asset_update_key) VALUES (null);
+INSERT INTO last_synced_key (last_synced_asset_update_key) VALUES (null);

migrations/9_insert_fungible_idx.sql

@@ -0,0 +1,3 @@
+ALTER TABLE last_synced_key DROP CONSTRAINT only_one_row;
+INSERT INTO last_synced_key (id, last_synced_asset_update_key) VALUES (2, NULL);
+ALTER TABLE last_synced_key ADD CONSTRAINT only_two_rows CHECK (id IN (1, 2));

nft_ingester/Cargo.toml

@@ -128,4 +128,4 @@ name = "raw_backfiller"
 name = "synchronizer"
 
 [[bin]]
-name = "raw_backup"
+name = "raw_backup"

nft_ingester/src/api/synchronization_state_consistency.rs

@@ -1,5 +1,6 @@
 use interface::consistency_check::ConsistencyChecker;
 use jsonrpc_core::Call;
+use postgre_client::asset_index_client::AssetType;
 use postgre_client::storage_traits::AssetIndexStorage;
 use postgre_client::PgClient;
 use rocks_db::key_encoders::decode_u64x2_pubkey;
@@ -42,41 +43,53 @@ impl SynchronizationStateConsistencyChecker {
     pub(crate) async fn run(
         &self,
         tasks: Arc<Mutex<JoinSet<Result<(), JoinError>>>>,
-        mut rx: tokio::sync::broadcast::Receiver<()>,
+        rx: tokio::sync::broadcast::Receiver<()>,
         pg_client: Arc<PgClient>,
         rocks_db: Arc<Storage>,
         synchronization_api_threshold: u64,
     ) {
-        let overwhelm_seq_gap_clone = self.overwhelm_seq_gap.clone();
-        tasks.lock().await.spawn(async move {
-            while rx.is_empty() {
-                let Ok(Some(index_seq)) = pg_client.fetch_last_synced_id().await else {
-                    continue;
-                };
-                let Ok(decoded_index_update_key) = decode_u64x2_pubkey(index_seq) else {
-                    continue;
-                };
-                let Ok(Some(primary_update_key)) = rocks_db.last_known_asset_updated_key() else {
-                    continue;
-                };
+        let asset_types = [AssetType::Fungible, AssetType::NonFungible];
 
-                overwhelm_seq_gap_clone.store(
-                    primary_update_key
-                        .seq
-                        .saturating_sub(decoded_index_update_key.seq)
-                        >= synchronization_api_threshold,
-                    Ordering::SeqCst,
-                );
-                tokio::select! {
-                    _ = tokio::time::sleep(Duration::from_secs(CATCH_UP_SEQUENCES_TIMEOUT_SEC))=> {},
-                    _ = rx.recv() => {
-                        info!("Received stop signal, stopping SynchronizationStateConsistencyChecker...");
-                        return Ok(());
+        for asset_type in asset_types {
+            let overwhelm_seq_gap = self.overwhelm_seq_gap.clone();
+            let pg_client = pg_client.clone();
+            let rocks_db = rocks_db.clone();
+            let mut rx = rx.resubscribe();
+            tasks.lock().await.spawn(async move {
+                while rx.is_empty() {
+                    let Ok(Some(index_seq)) = pg_client.fetch_last_synced_id(asset_type).await else {
+                        continue;
+                    };
+                    let Ok(decoded_index_update_key) = decode_u64x2_pubkey(index_seq) else {
+                        continue;
+                    };
+
+                    let last_known_updated_asset = match asset_type {
+                        AssetType::NonFungible => rocks_db.last_known_non_fungible_asset_updated_key(),
+                        AssetType::Fungible => rocks_db.last_known_fungible_asset_updated_key(),
+                    };
+                    let Ok(Some(primary_update_key)) = last_known_updated_asset else {
+                        continue;
+                    };
+
+                    overwhelm_seq_gap.store(
+                        primary_update_key
+                            .seq
+                            .saturating_sub(decoded_index_update_key.seq)
+                            >= synchronization_api_threshold,
+                        Ordering::SeqCst,
+                    );
+                    tokio::select! {
+                        _ = tokio::time::sleep(Duration::from_secs(CATCH_UP_SEQUENCES_TIMEOUT_SEC))=> {},
+                        _ = rx.recv() => {
+                            info!("Received stop signal, stopping SynchronizationStateConsistencyChecker...");
+                            return Ok(());
+                        }
                     }
                 }
-            }
-            Ok(())
-        });
+                Ok(())
+            });
+        }
     }
 }
 

nft_ingester/src/bin/ingester/main.rs

@@ -2,6 +2,7 @@ use arweave_rs::consts::ARWEAVE_BASE_URL;
 use arweave_rs::Arweave;
 use nft_ingester::batch_mint::batch_mint_persister::{BatchMintDownloaderForPersister, BatchMintPersister};
 use nft_ingester::scheduler::Scheduler;
+use postgre_client::asset_index_client::AssetType;
 use postgre_client::PG_MIGRATIONS_PATH;
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -580,15 +581,28 @@ pub async fn main() -> Result<(), IngesterError> {
         };
     }
 
+    let synchronizer = Arc::new(synchronizer);
     if !config.disable_synchronizer {
-        let rx = shutdown_rx.resubscribe();
-        mutexed_tasks.lock().await.spawn(async move {
-            synchronizer
-                .run(&rx, config.dump_sync_threshold, Duration::from_secs(5))
-                .await;
-
-            Ok(())
-        });
+        [AssetType::Fungible, AssetType::NonFungible]
+            .into_iter()
+            .for_each(|asset_type| {
+                let rx = shutdown_rx.resubscribe();
+                let synchronizer = synchronizer.clone();
+                tokio::spawn(async move {
+                    match asset_type {
+                        AssetType::NonFungible => {
+                            synchronizer
+                                .non_fungible_run(&rx, config.dump_sync_threshold, Duration::from_secs(5))
+                                .await
+                        }
+                        AssetType::Fungible => {
+                            synchronizer
+                                .fungible_run(&rx, config.dump_sync_threshold, Duration::from_secs(5))
+                                .await
+                        }
+                    }
+                });
+            })
     }
     // setup dependencies for grpc server
     let uc = AssetStreamer::new(config.peer_grpc_max_gap_slots, primary_rocks_storage.clone());

nft_ingester/src/bin/synchronizer/main.rs

@@ -6,6 +6,7 @@ use nft_ingester::config::{
 use nft_ingester::error::IngesterError;
 use nft_ingester::index_syncronizer::Synchronizer;
 use nft_ingester::init::{graceful_stop, init_index_storage_with_migration};
+use postgre_client::asset_index_client::AssetType;
 use postgre_client::PG_MIGRATIONS_PATH;
 use prometheus_client::registry::Registry;
 
@@ -111,7 +112,7 @@ pub async fn main() -> Result<(), IngesterError> {
         Ok(())
     });
 
-    let synchronizer = Synchronizer::new(
+    let synchronizer = Arc::new(Synchronizer::new(
         rocks_storage.clone(),
         index_storage.clone(),
         index_storage.clone(),
@@ -120,7 +121,7 @@ pub async fn main() -> Result<(), IngesterError> {
         metrics.clone(),
         config.parallel_tasks,
         config.run_temp_sync_during_dump,
-    );
+    ));
 
     if let Err(e) = rocks_storage.db.try_catch_up_with_primary() {
         tracing::error!("Sync rocksdb error: {}", e);
@@ -133,21 +134,42 @@ pub async fn main() -> Result<(), IngesterError> {
         if let Err(e) = rocks_storage.db.try_catch_up_with_primary() {
             tracing::error!("Sync rocksdb error: {}", e);
         }
-        let res = synchronizer
-            .synchronize_asset_indexes(&shutdown_rx, config.dump_sync_threshold)
-            .await;
-        match res {
-            Ok(_) => {
-                tracing::info!("Synchronization finished successfully");
-            }
-            Err(e) => {
-                tracing::error!("Synchronization failed: {:?}", e);
-            }
+
+        for asset_type in [AssetType::Fungible, AssetType::NonFungible] {
+            let synchronizer = synchronizer.clone();
+            let shutdown_rx = shutdown_rx.resubscribe();
+
+            tokio::spawn(async move {
+                let result = match asset_type {
+                    AssetType::NonFungible => {
+                        synchronizer
+                            .synchronize_non_fungible_asset_indexes(
+                                &shutdown_rx,
+                                config.dump_sync_threshold,
+                            )
+                            .await
+                    }
+                    AssetType::Fungible => {
+                        synchronizer
+                            .synchronize_fungible_asset_indexes(
+                                &shutdown_rx,
+                                config.dump_sync_threshold,
+                            )
+                            .await
+                    }
+                };
+
+                if let Err(e) = result {
+                    tracing::error!("Synchronization failed: {:?}", e);
+                } else {
+                    tracing::info!("Synchronization finished successfully");
+                }
+                tokio::time::sleep(tokio::time::Duration::from_secs(
+                    config.timeout_between_syncs_sec,
+                ))
+                .await;
+            });
         }
-        tokio::time::sleep(tokio::time::Duration::from_secs(
-            config.timeout_between_syncs_sec,
-        ))
-        .await;
     }
     Ok(())
 }