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

entities/src/enums.rs

@@ -400,3 +400,11 @@ pub enum TokenType {
     CompressedNFT,
     All,
 }
+
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub enum AssetType {
+    NonFungible = 1,
+    Fungible = 2,
+}
+
+pub const ASSET_TYPES: [AssetType; 2] = [AssetType::Fungible, AssetType::NonFungible];

entities/src/models.rs

@@ -73,6 +73,16 @@ 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 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/backfilling_state_consistency.rs

@@ -1,4 +1,5 @@
 use crate::api::synchronization_state_consistency::CATCH_UP_SEQUENCES_TIMEOUT_SEC;
+use entities::enums::{AssetType, ASSET_TYPES};
 use interface::consistency_check::ConsistencyChecker;
 use jsonrpc_core::Call;
 use rocks_db::Storage;
@@ -10,27 +11,35 @@ use tokio::task::{JoinError, JoinSet};
 use tracing::info;
 
 pub struct BackfillingStateConsistencyChecker {
-    overwhelm_backfill_gap: Arc<AtomicBool>,
+    overwhelm_fungible_backfill_gap: Arc<AtomicBool>,
+    overwhelm_non_backfill_gap: Arc<AtomicBool>,
 }
 
 impl BackfillingStateConsistencyChecker {
     pub(crate) fn new() -> Self {
         Self {
-            overwhelm_backfill_gap: Arc::new(AtomicBool::new(false)),
+            overwhelm_fungible_backfill_gap: Arc::new(AtomicBool::new(false)),
+            overwhelm_non_backfill_gap: Arc::new(AtomicBool::new(false)),
         }
     }
 
     pub(crate) async fn run(
         &self,
         tasks: Arc<Mutex<JoinSet<Result<(), JoinError>>>>,
-        mut rx: tokio::sync::broadcast::Receiver<()>,
+        rx: tokio::sync::broadcast::Receiver<()>,
         rocks_db: Arc<Storage>,
         consistence_backfilling_slots_threshold: u64,
     ) {
-        let overwhelm_backfill_gap_clone = self.overwhelm_backfill_gap.clone();
-        tasks.lock().await.spawn(async move {
+        for asset_type in ASSET_TYPES {
+            let rocks_db = rocks_db.clone();
+            let mut rx = rx.resubscribe();
+            let overwhelm_backfill_gap = match asset_type {
+                AssetType::NonFungible => self.overwhelm_non_backfill_gap.clone(),
+                AssetType::Fungible => self.overwhelm_fungible_backfill_gap.clone(),
+            };
+            tasks.lock().await.spawn(async move {
             while rx.is_empty() {
-                overwhelm_backfill_gap_clone.store(
+                overwhelm_backfill_gap.store(
                     rocks_db.bubblegum_slots.iter_start().count().saturating_add(rocks_db.ingestable_slots.iter_start().count())
                         >= consistence_backfilling_slots_threshold as usize,
                     Ordering::Relaxed,
@@ -45,11 +54,13 @@ impl BackfillingStateConsistencyChecker {
             }
             Ok(())
         });
+        }
     }
 }
 
 impl ConsistencyChecker for BackfillingStateConsistencyChecker {
     fn should_cancel_request(&self, _call: &Call) -> bool {
-        self.overwhelm_backfill_gap.load(Ordering::Relaxed)
+        self.overwhelm_non_backfill_gap.load(Ordering::Relaxed)
+            && self.overwhelm_fungible_backfill_gap.load(Ordering::Relaxed)
     }
 }

nft_ingester/src/api/synchronization_state_consistency.rs

@@ -1,3 +1,4 @@
+use entities::enums::{AssetType, ASSET_TYPES};
 use interface::consistency_check::ConsistencyChecker;
 use jsonrpc_core::Call;
 use postgre_client::storage_traits::AssetIndexStorage;
@@ -29,60 +30,77 @@ const INDEX_STORAGE_DEPENDS_METHODS: &[&str] = &[
 ];
 
 pub struct SynchronizationStateConsistencyChecker {
-    overwhelm_seq_gap: Arc<AtomicBool>,
+    overwhelm_non_fungible_seq_gap: Arc<AtomicBool>,
+    overwhelm_fungible_seq_gap: Arc<AtomicBool>,
 }
 
 impl SynchronizationStateConsistencyChecker {
     pub(crate) fn new() -> Self {
         Self {
-            overwhelm_seq_gap: Arc::new(AtomicBool::new(false)),
+            overwhelm_non_fungible_seq_gap: Arc::new(AtomicBool::new(false)),
+            overwhelm_fungible_seq_gap: Arc::new(AtomicBool::new(false)),
         }
     }
 
     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;
-                };
+        for asset_type in ASSET_TYPES {
+            let overwhelm_seq_gap = match asset_type {
+                AssetType::NonFungible => self.overwhelm_non_fungible_seq_gap.clone(),
+                AssetType::Fungible => self.overwhelm_fungible_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;
+                    };
 
-                overwhelm_seq_gap_clone.store(
-                    primary_update_key
-                        .seq
-                        .saturating_sub(decoded_index_update_key.seq)
-                        >= synchronization_api_threshold,
-                    Ordering::Relaxed,
-                );
-                tokio::select! {
-                    _ = tokio::time::sleep(Duration::from_secs(CATCH_UP_SEQUENCES_TIMEOUT_SEC))=> {},
-                    _ = rx.recv() => {
-                        info!("Received stop signal, stopping SynchronizationStateConsistencyChecker...");
-                        return Ok(());
+                    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::Relaxed,
+                    );
+                    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(())
+            });
+        }
     }
 }
 
 impl ConsistencyChecker for SynchronizationStateConsistencyChecker {
     fn should_cancel_request(&self, call: &Call) -> bool {
-        if !self.overwhelm_seq_gap.load(Ordering::Relaxed) {
+        if !&self.overwhelm_non_fungible_seq_gap.load(Ordering::Relaxed)
+            || !&self.overwhelm_fungible_seq_gap.load(Ordering::Relaxed)
+        {
             return false;
         }
 

nft_ingester/src/bin/ingester/main.rs

@@ -1,8 +1,10 @@
 use arweave_rs::consts::ARWEAVE_BASE_URL;
 use arweave_rs::Arweave;
+use entities::enums::{AssetType, ASSET_TYPES};
 use nft_ingester::batch_mint::batch_mint_persister::{BatchMintDownloaderForPersister, BatchMintPersister};
 use nft_ingester::scheduler::Scheduler;
 use postgre_client::PG_MIGRATIONS_PATH;
+use std::panic;
 use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -121,7 +123,7 @@ pub async fn main() -> Result<(), IngesterError> {
         .await?,
     );
 
-    let synchronizer = Synchronizer::new(
+    let synchronizer = Arc::new(Synchronizer::new(
         primary_rocks_storage.clone(),
         index_pg_storage.clone(),
         index_pg_storage.clone(),
@@ -130,11 +132,33 @@ pub async fn main() -> Result<(), IngesterError> {
         metrics_state.synchronizer_metrics.clone(),
         config.synchronizer_parallel_tasks,
         config.run_temp_sync_during_dump,
-    );
+    ));
 
     if config.run_dump_synchronize_on_start {
         info!("Running dump synchronizer on start!");
-        synchronizer.full_syncronize(&shutdown_rx.resubscribe()).await?;
+        for asset_type in ASSET_TYPES {
+            let synchronizer = synchronizer.clone();
+            let shutdown_rx = shutdown_rx.resubscribe();
+
+            mutexed_tasks.lock().await.spawn(async move {
+                if let Err(e) = synchronizer
+                    .full_syncronize(&shutdown_rx.resubscribe(), asset_type)
+                    .await
+                {
+                    error!("Failed to syncronize on {:?} with error {}", asset_type, e);
+                    panic!("Failed to syncronize on {:?} with error {}", asset_type, e);
+                }
+
+                Ok(())
+            });
+        }
+    }
+    while let Some(res) = mutexed_tasks.lock().await.join_next().await {
+        match res {
+            Ok(_) => {}
+            Err(err) if err.is_panic() => panic::resume_unwind(err.into_panic()),
+            Err(err) => panic!("{err}"),
+        }
     }
 
     // setup receiver
@@ -581,15 +605,37 @@ pub async fn main() -> Result<(), IngesterError> {
     }
 
     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;
+        let synchronizer = Arc::new(synchronizer);
 
-            Ok(())
-        });
+        for asset_type in ASSET_TYPES {
+            let rx = shutdown_rx.resubscribe();
+            let synchronizer = synchronizer.clone();
+            mutexed_tasks.lock().await.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
+                    }
+                }
+
+                Ok(())
+            });
+        }
+    }
+    while let Some(res) = mutexed_tasks.lock().await.join_next().await {
+        match res {
+            Ok(_) => {}
+            Err(err) if err.is_panic() => panic::resume_unwind(err.into_panic()),
+            Err(err) => panic!("{err}"),
+        }
     }
+
     // setup dependencies for grpc server
     let uc = AssetStreamer::new(config.peer_grpc_max_gap_slots, primary_rocks_storage.clone());
     let bs = BlocksStreamer::new(config.peer_grpc_max_gap_slots, primary_rocks_storage.clone());