feat: determinate gape slot range

nft_ingester/src/bin/ingester/main.rs

@@ -1,5 +1,6 @@
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 use clap::Parser;
 use log::{error, info};
@@ -26,6 +27,7 @@ use nft_ingester::{config::init_logger, error::IngesterError};
 use postgre_client::PgClient;
 use rocks_db::backup_service::BackupService;
 use rocks_db::errors::BackupServiceError;
+use rocks_db::storage_traits::AssetSlotStorage;
 use rocks_db::{backup_service, Storage};
 
 mod backfiller;
@@ -138,6 +140,10 @@ pub async fn main() -> Result<(), IngesterError> {
     .unwrap();
 
     let rocks_storage = Arc::new(storage);
+    let newest_restored_slot = rocks_storage
+        .last_saved_slot()
+        .unwrap()
+        .ok_or(IngesterError::EmptyDataBase)?;
 
     // start backup service
     let backup_cfg = backup_service::load_config()?;
@@ -195,6 +201,28 @@ pub async fn main() -> Result<(), IngesterError> {
         }));
     }
 
+    let first_processed_slot = Arc::new(AtomicU64::new(0));
+    let first_processed_slot_clone = first_processed_slot.clone();
+    let cloned_rocks_storage = rocks_storage.clone();
+    let cloned_keep_running = keep_running.clone();
+    tasks.spawn(tokio::spawn(async move {
+        while cloned_keep_running.load(Ordering::SeqCst) {
+            let slot = cloned_rocks_storage
+                .last_saved_slot()
+                .unwrap_or({
+                    cloned_keep_running.store(false, Ordering::SeqCst);
+                    None
+                }) // If error returned from DB - stop all services
+                .unwrap(); // DB is not empty, so panic only in error case
+            if slot != newest_restored_slot {
+                first_processed_slot_clone.store(slot, Ordering::SeqCst);
+                break;
+            }
+
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }));
+
     let cloned_keep_running = keep_running.clone();
     let cloned_rocks_storage = rocks_storage.clone();
     tasks.spawn(tokio::spawn(async move {

nft_ingester/src/error/mod.rs

@@ -92,6 +92,8 @@ pub enum IngesterError {
     TransactionNotProcessedError(String),
     #[error("backup service {0}")]
     BackupError(String),
+    #[error("Trying to run services with empty DB. Please restart app with added --restore-rocks-db flag")]
+    EmptyDataBase,
 }
 
 impl From<reqwest::Error> for IngesterError {

rocks-db/src/batch_client.rs

@@ -3,10 +3,10 @@ use std::collections::{HashMap, HashSet};
 use async_trait::async_trait;
 use solana_sdk::pubkey::Pubkey;
 
-use crate::asset::AssetsUpdateIdx;
+use crate::asset::{AssetsUpdateIdx, SlotAssetIdx};
 use crate::column::TypedColumn;
 use crate::key_encoders::{decode_u64x2_pubkey, encode_u64x2_pubkey};
-use crate::storage_traits::{AssetIndexReader, AssetUpdateIndexStorage};
+use crate::storage_traits::{AssetIndexReader, AssetSlotStorage, AssetUpdateIndexStorage};
 use crate::{Result, Storage};
 use entities::models::AssetIndex;
 
@@ -190,3 +190,16 @@ impl AssetIndexReader for Storage {
         Ok(asset_indexes)
     }
 }
+
+impl AssetSlotStorage for Storage {
+    fn last_saved_slot(&self) -> Result<Option<u64>> {
+        let mut iter = self.slot_asset_idx.iter_end();
+        if let Some(pair) = iter.next() {
+            let (last_key, _) = pair?;
+            let (slot, _) = SlotAssetIdx::decode_key(last_key.to_vec())?;
+            return Ok(Some(slot));
+        }
+
+        Ok(None)
+    }
+}

rocks-db/src/column.rs

@@ -143,12 +143,6 @@ pub mod columns {
         pub mint_authority: Option<Pubkey>,
         pub freeze_authority: Option<Pubkey>,
     }
-
-    #[derive(Debug, Serialize, Deserialize)]
-    pub struct TokenAccountOwnerIdx {}
-
-    #[derive(Debug, Serialize, Deserialize)]
-    pub struct TokenAccountMintIdx {}
 }
 
 impl TypedColumn for columns::TokenAccount {
@@ -204,47 +198,3 @@ impl columns::TokenAccount {
         Some(result)
     }
 }
-
-impl TypedColumn for columns::TokenAccountOwnerIdx {
-    const NAME: &'static str = "ACCOUNTS_TOKEN_OWNER_IDX";
-
-    type KeyType = (Pubkey, Pubkey);
-    type ValueType = Self;
-
-    fn encode_key((owner, pubkey): (Pubkey, Pubkey)) -> Vec<u8> {
-        let mut key = vec![0; 32 + 32]; // size_of Pubkey + size_of Pubkey
-        key[0..32].clone_from_slice(&owner.as_ref()[0..32]);
-        key[32..64].clone_from_slice(&pubkey.as_ref()[0..32]);
-
-        key
-    }
-
-    fn decode_key(bytes: Vec<u8>) -> Result<Self::KeyType> {
-        let owner = Pubkey::try_from(&bytes[0..32])?;
-        let pubkey = Pubkey::try_from(&bytes[32..64])?;
-
-        Ok((owner, pubkey))
-    }
-}
-
-impl TypedColumn for columns::TokenAccountMintIdx {
-    const NAME: &'static str = "ACCOUNTS_TOKEN_MINT_IDX";
-
-    type KeyType = (Pubkey, Pubkey);
-    type ValueType = Self;
-
-    fn encode_key((owner, pubkey): (Pubkey, Pubkey)) -> Vec<u8> {
-        let mut key = vec![0; 32 + 32]; // size_of Pubkey + size_of Pubkey
-        key[0..32].clone_from_slice(&owner.as_ref()[0..32]);
-        key[32..64].clone_from_slice(&pubkey.as_ref()[0..32]);
-
-        key
-    }
-
-    fn decode_key(bytes: Vec<u8>) -> Result<Self::KeyType> {
-        let owner = Pubkey::try_from(&bytes[0..32])?;
-        let pubkey = Pubkey::try_from(&bytes[32..64])?;
-
-        Ok((owner, pubkey))
-    }
-}

rocks-db/src/lib.rs

@@ -37,9 +37,6 @@ pub struct Storage {
     pub cl_items: Column<cl_items::ClItem>,
     pub cl_leafs: Column<cl_items::ClLeaf>,
     pub bubblegum_slots: Column<bubblegum_slots::BubblegumSlots>,
-    pub token_accounts: Column<columns::TokenAccount>,
-    pub account_token_owner_idx: Column<columns::TokenAccountOwnerIdx>,
-    pub account_token_mint_idx: Column<columns::TokenAccountMintIdx>,
     pub db: Arc<DB>,
     pub assets_update_idx: Column<AssetsUpdateIdx>,
     pub slot_asset_idx: Column<SlotAssetIdx>,
@@ -62,9 +59,6 @@ impl Storage {
                 Self::new_cf_descriptor::<cl_items::ClItem>(),
                 Self::new_cf_descriptor::<cl_items::ClLeaf>(),
                 Self::new_cf_descriptor::<bubblegum_slots::BubblegumSlots>(),
-                Self::new_cf_descriptor::<columns::TokenAccount>(),
-                Self::new_cf_descriptor::<columns::TokenAccountOwnerIdx>(),
-                Self::new_cf_descriptor::<columns::TokenAccountMintIdx>(),
                 Self::new_cf_descriptor::<asset::AssetsUpdateIdx>(),
                 Self::new_cf_descriptor::<asset::SlotAssetIdx>(),
             ],
@@ -83,9 +77,6 @@ impl Storage {
 
         let bubblegum_slots = Self::column(db.clone());
 
-        let token_accounts = Self::column(db.clone());
-        let account_token_owner_idx = Self::column(db.clone());
-        let account_token_mint_idx = Self::column(db.clone());
         let assets_update_idx = Self::column(db.clone());
         let slot_asset_idx = Self::column(db.clone());
 
@@ -100,9 +91,6 @@ impl Storage {
             cl_items,
             cl_leafs,
             bubblegum_slots,
-            token_accounts,
-            account_token_owner_idx,
-            account_token_mint_idx,
             db,
             assets_update_idx,
             slot_asset_idx,

rocks-db/src/storage_traits.rs

@@ -72,3 +72,8 @@ impl AssetIndexReader for MockAssetIndexStorage {
 impl AssetIndexStorage for MockAssetIndexStorage {}
 
 impl AssetIndexStorage for Storage {}
+
+#[automock]
+pub trait AssetSlotStorage {
+    fn last_saved_slot(&self) -> Result<Option<u64>>;
+}