[data ingestion] for lagging secondary only update local progress sta…

…te (#18927)

crates/sui-data-ingestion-core/src/progress_store/mod.rs

@@ -38,15 +38,12 @@ impl<P: ProgressStore> ProgressStore for ProgressStoreWrapper<P> {
         task_name: String,
         checkpoint_number: CheckpointSequenceNumber,
     ) -> Result<()> {
-        let last_saved = self.load(task_name.clone()).await?;
-        if checkpoint_number > last_saved {
+        if checkpoint_number > self.load(task_name.clone()).await? {
             self.progress_store
                 .save(task_name.clone(), checkpoint_number)
                 .await?;
-            self.pending_state.insert(task_name, checkpoint_number);
-        } else {
-            self.pending_state.insert(task_name, last_saved);
         }
+        self.pending_state.insert(task_name, checkpoint_number);
         Ok(())
     }
 }

crates/sui-data-ingestion-core/src/reader.rs

@@ -11,7 +11,6 @@ use notify::RecursiveMode;
 use notify::Watcher;
 use object_store::path::Path;
 use object_store::ObjectStore;
-use std::cmp::max;
 use std::collections::BTreeMap;
 use std::ffi::OsString;
 use std::fs;
@@ -293,7 +292,6 @@ impl CheckpointReader {
         info!("cleaning processed files, watermark is {}", watermark);
         self.data_limiter.gc(watermark);
         self.last_pruned_watermark = watermark;
-        self.current_checkpoint_number = max(self.current_checkpoint_number, watermark);
         for entry in fs::read_dir(self.path.clone())? {
             let entry = entry?;
             let filename = entry.file_name();