feat: determinate gape slot range

nft_ingester/src/bin/ingester/main.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 
 use clap::Parser;
@@ -26,6 +26,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;
@@ -85,7 +86,11 @@ pub async fn main() -> Result<(), IngesterError> {
     let buffer = Arc::new(Buffer::new());
 
     // setup receiver
-    let message_handler = Arc::new(MessageHandler::new(buffer.clone()));
+    let first_processed_slot = Arc::new(AtomicU64::new(0));
+    let message_handler = Arc::new(MessageHandler::new(
+        buffer.clone(),
+        first_processed_slot.clone(),
+    ));
 
     let geyser_tcp_receiver = TcpReceiver::new(
         message_handler.clone(),
@@ -138,6 +143,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()?;

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 {

nft_ingester/src/message_handler.rs

@@ -1,3 +1,4 @@
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::{str::FromStr, sync::Arc};
 
 use blockbuster::programs::token_account::TokenProgramAccount;
@@ -33,17 +34,19 @@ pub struct MessageHandler {
     buffer: Arc<Buffer>,
     token_acc_parser: Arc<TokenAccountParser>,
     mplx_acc_parser: Arc<TokenMetadataParser>,
+    first_processed_slot: Arc<AtomicU64>,
 }
 
 impl MessageHandler {
-    pub fn new(buffer: Arc<Buffer>) -> Self {
+    pub fn new(buffer: Arc<Buffer>, first_processed_slot: Arc<AtomicU64>) -> Self {
         let token_acc_parser = Arc::new(TokenAccountParser {});
         let mplx_acc_parser = Arc::new(TokenMetadataParser {});
 
         Self {
             buffer,
             token_acc_parser,
             mplx_acc_parser,
+            first_processed_slot,
         }
     }
 
@@ -69,11 +72,21 @@ impl MessageHandler {
             _ => {}
         }
     }
+    fn store_first_processed_slot(&self, slot: u64) {
+        if self.first_processed_slot.load(Ordering::SeqCst) != 0 {
+            // slot already stored
+            return;
+        }
+
+        self.first_processed_slot.store(slot, Ordering::SeqCst)
+    }
 
     async fn handle_account(&self, data: Vec<u8>) -> Result<(), IngesterError> {
         let account_update =
             utils::flatbuffer::account_info_generated::account_info::root_as_account_info(&data)?;
 
+        self.store_first_processed_slot(account_update.slot());
+
         let account_info_bytes = map_account_info_fb_bytes(account_update)?;
 
         let account_info = plerkle_serialization::root_as_account_info(account_info_bytes.as_ref())

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>(),
             ],
         )?);
@@ -82,9 +76,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());
 
@@ -99,9 +90,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>>;
+}