Pass last updated key to the cleaup function instead of using naive i…

…teration
metaplex-foundation · Nov 20, 2024 · 037fea8 · 037fea8
1 parent dbe9e3f
commit 037fea8
Show file tree

Hide file tree

Showing 6 changed files with 50 additions and 17 deletions.
diff --git a/nft_ingester/src/index_syncronizer.rs b/nft_ingester/src/index_syncronizer.rs
@@ -200,7 +200,11 @@ where
             SyncStatus::NoSyncRequired => {}
         }
 
-        self.clean_syncronized_idxs(asset_type)
+        if let Some(encoded_key) = self.index_storage.fetch_last_synced_id(asset_type).await? {
+            self.clean_syncronized_idxs(asset_type, encoded_key)?;
+        }
+
+        Ok(())
     }
 
     pub async fn synchronize_fungible_asset_indexes(
@@ -226,12 +230,20 @@ where
             }
             SyncStatus::NoSyncRequired => {}
         }
+        if let Some(encoded_key) = self.index_storage.fetch_last_synced_id(asset_type).await? {
+            self.clean_syncronized_idxs(asset_type, encoded_key)?;
+        }
 
-        self.clean_syncronized_idxs(asset_type)
+        Ok(())
     }
 
-    pub fn clean_syncronized_idxs(&self, asset_type: AssetType) -> Result<(), IngesterError> {
-        self.primary_storage.clean_syncronized_idxs(asset_type)?;
+    pub fn clean_syncronized_idxs(
+        &self,
+        asset_type: AssetType,
+        last_synced_key: Vec<u8>,
+    ) -> Result<(), IngesterError> {
+        self.primary_storage
+            .clean_syncronized_idxs(asset_type, last_synced_key)?;
 
         Ok(())
     }

diff --git a/nft_ingester/src/processors/account_based/token_updates_processor.rs b/nft_ingester/src/processors/account_based/token_updates_processor.rs
@@ -188,10 +188,13 @@ impl TokenAccountsProcessor {
         &self,
         storage: &mut BatchSaveStorage,
         asset_type: AssetType,
+        last_synced_key: Vec<u8>,
     ) -> Result<(), StorageError> {
         self.finalize_processing(
             storage,
-            |storage: &mut BatchSaveStorage| storage.clean_syncronized_idxs(asset_type),
+            |storage: &mut BatchSaveStorage| {
+                storage.clean_syncronized_idxs(asset_type, last_synced_key.clone())
+            },
             "clean_syncronized_idxs",
         )
     }

diff --git a/rocks-db/src/asset_client.rs b/rocks-db/src/asset_client.rs
@@ -114,19 +114,23 @@ impl Storage {
         Ok(())
     }
 
-    pub fn clean_syncronized_idxs_with_batch(&self, asset_type: AssetType) -> Result<()> {
+    pub fn clean_syncronized_idxs_with_batch(
+        &self,
+        asset_type: AssetType,
+        last_synced_key: Vec<u8>,
+    ) -> Result<()> {
         let (mut iter, cf) = match asset_type {
             AssetType::Fungible => {
                 let cf = self.fungible_assets_update_idx.handle();
-                // TODO: use regular iter to avoid data inconcistency
-                let iter = self.fungible_assets_update_idx.iter_end();
+                let iter = self
+                    .fungible_assets_update_idx
+                    .iter_reverse(last_synced_key);
 
                 (iter, cf)
             }
             AssetType::NonFungible => {
                 let cf = self.assets_update_idx.handle();
-                // TODO: use regular iter to avoid data inconcistency
-                let iter = self.assets_update_idx.iter_end();
+                let iter = self.assets_update_idx.iter_reverse(last_synced_key);
 
                 (iter, cf)
             }

diff --git a/rocks-db/src/batch_client.rs b/rocks-db/src/batch_client.rs
@@ -184,8 +184,12 @@ impl AssetUpdateIndexStorage for Storage {
         Ok((unique_pubkeys, last_key))
     }
 
-    fn clean_syncronized_idxs(&self, asset_type: AssetType) -> Result<()> {
-        self.clean_syncronized_idxs_with_batch(asset_type)
+    fn clean_syncronized_idxs(
+        &self,
+        asset_type: AssetType,
+        last_synced_key: Vec<u8>,
+    ) -> Result<()> {
+        self.clean_syncronized_idxs_with_batch(asset_type, last_synced_key)
     }
 }
 

diff --git a/rocks-db/src/batch_savers.rs b/rocks-db/src/batch_savers.rs
@@ -212,8 +212,13 @@ impl BatchSaveStorage {
         )
     }
 
-    pub fn clean_syncronized_idxs(&self, asset_type: AssetType) -> Result<()> {
-        self.storage.clean_syncronized_idxs_with_batch(asset_type)
+    pub fn clean_syncronized_idxs(
+        &self,
+        asset_type: AssetType,
+        last_synced_key: Vec<u8>,
+    ) -> Result<()> {
+        self.storage
+            .clean_syncronized_idxs_with_batch(asset_type, last_synced_key)
     }
 
     pub fn get_authority(&self, address: Pubkey) -> Pubkey {

diff --git a/rocks-db/src/storage_traits.rs b/rocks-db/src/storage_traits.rs
@@ -46,7 +46,8 @@ pub trait AssetUpdateIndexStorage {
         skip_keys: Option<HashSet<Pubkey>>,
     ) -> Result<(HashSet<Pubkey>, Option<AssetUpdatedKey>)>;
 
-    fn clean_syncronized_idxs(&self, asset_type: AssetType) -> Result<()>;
+    fn clean_syncronized_idxs(&self, asset_type: AssetType, last_synced_key: Vec<u8>)
+        -> Result<()>;
 }
 
 #[automock]
@@ -128,9 +129,13 @@ impl AssetUpdateIndexStorage for MockAssetIndexStorage {
             .fetch_fungible_asset_updated_keys(from, up_to, limit, skip_keys)
     }
 
-    fn clean_syncronized_idxs(&self, asset_type: AssetType) -> Result<()> {
+    fn clean_syncronized_idxs(
+        &self,
+        asset_type: AssetType,
+        last_synced_key: Vec<u8>,
+    ) -> Result<()> {
         self.mock_update_index_storage
-            .clean_syncronized_idxs(asset_type)
+            .clean_syncronized_idxs(asset_type, last_synced_key)
     }
 }