feat(sync): update pending only when it's more advanced (#1288)

crates/papyrus_sync/src/lib.rs

@@ -882,6 +882,7 @@ async fn sync_pending_data<TPendingSource: PendingSourceTrait + Sync + Send>(
     pending_data: Arc<RwLock<PendingData>>,
     sleep_duration: Duration,
 ) -> Result<(), StateSyncError> {
+    *pending_data.write().await = PendingData::default();
     let txn = reader.begin_ro_txn()?;
     let header_marker = txn.get_header_marker()?;
     // TODO: Consider extracting this functionality to different а function.
@@ -898,9 +899,17 @@ async fn sync_pending_data<TPendingSource: PendingSourceTrait + Sync + Send>(
     loop {
         let new_pending_data = pending_source.get_pending_data().await?;
         if new_pending_data.block.parent_block_hash != latest_block_hash {
+            debug!("A new block was found. Stopping pending sync.");
             return Ok(());
         };
-        *pending_data.write().await = new_pending_data;
+        let current_pending_num_transactions = pending_data.read().await.block.transactions.len();
+        if current_pending_num_transactions == 0
+            || new_pending_data.block.transactions.len() > current_pending_num_transactions
+        {
+            debug!("Received new pending data.");
+            trace!("Pending data: {new_pending_data:#?}.");
+            *pending_data.write().await = new_pending_data;
+        }
         tokio::time::sleep(sleep_duration).await;
     }
 }

crates/papyrus_sync/src/sources/pending.rs

@@ -14,7 +14,6 @@ use starknet_client::reader::{
     StarknetReader,
 };
 use starknet_client::ClientCreationError;
-use tracing::{debug, trace};
 
 // TODO(dvir): add pending config.
 use super::central::CentralSourceConfig;
@@ -44,11 +43,7 @@ impl<TStarknetClient: StarknetReader + Send + Sync + 'static> PendingSourceTrait
 {
     async fn get_pending_data(&self) -> Result<PendingData, PendingError> {
         match self.starknet_client.pending_data().await {
-            Ok(Some(pending_data)) => {
-                debug!("Received new pending data.");
-                trace!("Pending data: {pending_data:#?}.");
-                Ok(pending_data)
-            }
+            Ok(Some(pending_data)) => Ok(pending_data),
             Ok(None) => Err(PendingError::PendingBlockNotFound),
             Err(err) => Err(PendingError::ClientError(Arc::new(err))),
         }