Merge ec21c6fce4 into b6972b97af

node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs

@@ -17,12 +17,9 @@ use std::sync::Arc;
 use std::time::Duration;
 use storage::log_store::{tx_store::BlockHashAndSubmissionIndex, Store};
 use task_executor::TaskExecutor;
-use tokio::{
-    sync::{
-        mpsc::{UnboundedReceiver, UnboundedSender},
-        RwLock,
-    },
-    time::Instant,
+use tokio::sync::{
+    mpsc::{UnboundedReceiver, UnboundedSender},
+    RwLock,
 };
 
 pub struct LogEntryFetcher {
@@ -252,10 +249,7 @@ impl LogEntryFetcher {
                             }) {
                                 Ok(event) => {
                                     if let Err(e) = recover_tx
-                                        .send(submission_event_to_transaction(
-                                            event,
-                                            log.block_number.expect("block number exist").as_u64(),
-                                        ))
+                                        .send(submission_event_to_transaction(event))
                                         .and_then(|_| match sync_progress {
                                             Some(b) => recover_tx.send(b),
                                             None => Ok(()),
@@ -291,14 +285,12 @@ impl LogEntryFetcher {
         executor: &TaskExecutor,
         block_hash_cache: Arc<RwLock<BTreeMap<u64, Option<BlockHashAndSubmissionIndex>>>>,
         watch_loop_wait_time_ms: u64,
-        mut watch_progress_rx: UnboundedReceiver<u64>,
     ) -> UnboundedReceiver<LogFetchProgress> {
         let (watch_tx, watch_rx) = tokio::sync::mpsc::unbounded_channel();
         let contract = ZgsFlow::new(self.contract_address, self.provider.clone());
         let provider = self.provider.clone();
         let confirmation_delay = self.confirmation_delay;
         let log_page_size = self.log_page_size;
-        let mut progress_reset_history = BTreeMap::new();
         executor.spawn(
             async move {
                 debug!("start_watch starts, start={}", start_block_number);
@@ -306,13 +298,6 @@ impl LogEntryFetcher {
                 let mut parent_block_hash = parent_block_hash;
 
                 loop {
-                    check_watch_process(
-                        &mut watch_progress_rx,
-                        &mut progress,
-                        &mut progress_reset_history,
-                        watch_loop_wait_time_ms,
-                    );
-
                     match Self::watch_loop(
                         provider.as_ref(),
                         progress,
@@ -504,11 +489,8 @@ impl LogEntryFetcher {
                             first_submission_index = Some(submit_filter.submission_index.as_u64());
                         }
 
-                        log_events
-                            .push(submission_event_to_transaction(submit_filter, block_number));
+                        log_events.push(submission_event_to_transaction(submit_filter));
                     }
-
-                    info!("synced {} events", log_events.len());
                 }
 
                 let new_progress = if block.hash.is_some() && block.number.is_some() {
@@ -526,21 +508,12 @@ impl LogEntryFetcher {
                         return Ok(progress);
                     }
                 }
-
                 if let Some(p) = &new_progress {
                     if let Err(e) = watch_tx.send(LogFetchProgress::SyncedBlock(*p)) {
                         warn!("send LogFetchProgress::SyncedBlock failed: {:?}", e);
                         return Ok(progress);
                     } else {
-                        let mut cache = block_hash_cache.write().await;
-                        match cache.get(&p.0) {
-                            Some(Some(v))
-                                if v.block_hash == p.1
-                                    && v.first_submission_index == p.2.unwrap() => {}
-                            _ => {
-                                cache.insert(p.0, None);
-                            }
-                        }
+                        block_hash_cache.write().await.insert(p.0, None);
                     }
                 }
                 progress = new_progress;
@@ -555,59 +528,6 @@ impl LogEntryFetcher {
     }
 }
 
-fn check_watch_process(
-    watch_progress_rx: &mut UnboundedReceiver<u64>,
-    progress: &mut u64,
-    progress_reset_history: &mut BTreeMap<u64, (Instant, usize)>,
-    watch_loop_wait_time_ms: u64,
-) {
-    let mut min_received_progress = None;
-    while let Ok(v) = watch_progress_rx.try_recv() {
-        min_received_progress = match min_received_progress {
-            Some(min) if min > v => Some(v),
-            None => Some(v),
-            _ => min_received_progress,
-        };
-    }
-
-    if let Some(v) = min_received_progress {
-        if *progress <= v {
-            error!(
-                "received unexpected progress, current {}, received {}",
-                *progress, v
-            );
-            return;
-        }
-
-        let now = Instant::now();
-        match progress_reset_history.get_mut(&v) {
-            Some((last_update, counter)) => {
-                if *counter >= 3 {
-                    error!("maximum reset attempts have been reached.");
-                    watch_progress_rx.close();
-                    return;
-                }
-
-                if now.duration_since(*last_update)
-                    >= Duration::from_millis(watch_loop_wait_time_ms * 30)
-                {
-                    info!("reset to progress from {} to {}", *progress, v);
-                    *progress = v;
-                    *last_update = now;
-                    *counter += 1;
-                }
-            }
-            None => {
-                info!("reset to progress from {} to {}", *progress, v);
-                *progress = v;
-                progress_reset_history.insert(v, (now, 1usize));
-            }
-        }
-    }
-
-    progress_reset_history.retain(|k, _| k + 1000 >= *progress);
-}
-
 async fn revert_one_block(
     block_hash: H256,
     block_number: u64,
@@ -661,29 +581,26 @@ async fn revert_one_block(
 #[derive(Debug)]
 pub enum LogFetchProgress {
     SyncedBlock((u64, H256, Option<Option<u64>>)),
-    Transaction((Transaction, u64)),
+    Transaction(Transaction),
     Reverted(u64),
 }
 
-fn submission_event_to_transaction(e: SubmitFilter, block_number: u64) -> LogFetchProgress {
-    LogFetchProgress::Transaction((
-        Transaction {
-            stream_ids: vec![],
-            data: vec![],
-            data_merkle_root: nodes_to_root(&e.submission.nodes),
-            merkle_nodes: e
-                .submission
-                .nodes
-                .iter()
-                // the submission height is the height of the root node starting from height 0.
-                .map(|SubmissionNode { root, height }| (height.as_usize() + 1, root.into()))
-                .collect(),
-            start_entry_index: e.start_pos.as_u64(),
-            size: e.submission.length.as_u64(),
-            seq: e.submission_index.as_u64(),
-        },
-        block_number,
-    ))
+fn submission_event_to_transaction(e: SubmitFilter) -> LogFetchProgress {
+    LogFetchProgress::Transaction(Transaction {
+        stream_ids: vec![],
+        data: vec![],
+        data_merkle_root: nodes_to_root(&e.submission.nodes),
+        merkle_nodes: e
+            .submission
+            .nodes
+            .iter()
+            // the submission height is the height of the root node starting from height 0.
+            .map(|SubmissionNode { root, height }| (height.as_usize() + 1, root.into()))
+            .collect(),
+        start_entry_index: e.start_pos.as_u64(),
+        size: e.submission.length.as_u64(),
+        seq: e.submission_index.as_u64(),
+    })
 }
 
 fn nodes_to_root(node_list: &[SubmissionNode]) -> DataRoot {

node/log_entry_sync/src/sync_manager/mod.rs

@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
 use storage::log_store::{tx_store::BlockHashAndSubmissionIndex, Store};
 use task_executor::{ShutdownReason, TaskExecutor};
 use tokio::sync::broadcast;
-use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::mpsc::UnboundedReceiver;
 use tokio::sync::{oneshot, RwLock};
 
 const RETRY_WAIT_MS: u64 = 500;
@@ -103,7 +103,16 @@ impl LogSyncManager {
                     };
 
                     let (mut start_block_number, mut start_block_hash) =
-                        get_start_block_number_with_hash(&log_sync_manager).await?;
+                        match log_sync_manager.store.get_sync_progress()? {
+                            // No previous progress, so just use config.
+                            None => {
+                                let block_number = log_sync_manager.config.start_block_number;
+                                let block_hash =
+                                    log_sync_manager.get_block(block_number.into()).await?.1;
+                                (block_number, block_hash)
+                            }
+                            Some((block_number, block_hash)) => (block_number, block_hash),
+                        };
 
                     let (mut finalized_block_number, mut finalized_block_hash) =
                         match log_sync_manager.get_block(BlockNumber::Finalized).await {
@@ -137,7 +146,7 @@ impl LogSyncManager {
                             &executor_clone,
                             log_sync_manager.block_hash_cache.clone(),
                         );
-                        log_sync_manager.handle_data(reorg_rx, &None).await?;
+                        log_sync_manager.handle_data(reorg_rx).await?;
                         if let Some((block_number, block_hash)) =
                             log_sync_manager.store.get_sync_progress()?
                         {
@@ -214,20 +223,15 @@ impl LogSyncManager {
                         warn!("catch_up_end send fails, possibly auto_sync is not enabled");
                     }
 
-                    let (watch_progress_tx, watch_progress_rx) =
-                        tokio::sync::mpsc::unbounded_channel();
                     let watch_rx = log_sync_manager.log_fetcher.start_watch(
                         start_block_number,
                         parent_block_hash,
                         &executor_clone,
                         log_sync_manager.block_hash_cache.clone(),
                         log_sync_manager.config.watch_loop_wait_time_ms,
-                        watch_progress_rx,
                     );
                     // Syncing `watch_rx` is supposed to block forever.
-                    log_sync_manager
-                        .handle_data(watch_rx, &Some(watch_progress_tx))
-                        .await?;
+                    log_sync_manager.handle_data(watch_rx).await?;
                     Ok::<(), anyhow::Error>(())
                 },
             )
@@ -237,20 +241,20 @@ impl LogSyncManager {
         Ok((event_send_cloned, catch_up_end_receiver))
     }
 
-    async fn put_tx(&mut self, tx: Transaction) -> Option<bool> {
+    async fn put_tx(&mut self, tx: Transaction) -> bool {
         // We call this after process chain reorg, so the sequence number should match.
         match tx.seq.cmp(&self.next_tx_seq) {
-            std::cmp::Ordering::Less => Some(true),
+            std::cmp::Ordering::Less => true,
             std::cmp::Ordering::Equal => {
                 debug!("log entry sync get entry: {:?}", tx);
-                Some(self.put_tx_inner(tx).await)
+                self.put_tx_inner(tx).await
             }
             std::cmp::Ordering::Greater => {
                 error!(
                     "Unexpected transaction seq: next={} get={}",
                     self.next_tx_seq, tx.seq
                 );
-                None
+                false
             }
         }
     }
@@ -292,18 +296,7 @@ impl LogSyncManager {
         let _ = self.event_send.send(LogSyncEvent::Reverted { tx_seq });
     }
 
-    async fn handle_data(
-        &mut self,
-        mut rx: UnboundedReceiver<LogFetchProgress>,
-        watch_progress_tx: &Option<UnboundedSender<u64>>,
-    ) -> Result<()> {
-        let mut log_latest_block_number =
-            if let Some(block_number) = self.store.get_log_latest_block_number()? {
-                block_number
-            } else {
-                0
-            };
-
+    async fn handle_data(&mut self, mut rx: UnboundedReceiver<LogFetchProgress>) -> Result<()> {
         while let Some(data) = rx.recv().await {
             trace!("handle_data: data={:?}", data);
             match data {
@@ -343,30 +336,8 @@ impl LogSyncManager {
                         }
                     }
                 }
-                LogFetchProgress::Transaction((tx, block_number)) => {
-                    let mut stop = false;
-                    match self.put_tx(tx.clone()).await {
-                        Some(false) => stop = true,
-                        Some(true) => {
-                            if let Err(e) = self.store.put_log_latest_block_number(block_number) {
-                                warn!("failed to put log latest block number, error={:?}", e);
-                            }
-
-                            log_latest_block_number = block_number;
-                        }
-                        _ => {
-                            stop = true;
-                            if let Some(progress_tx) = watch_progress_tx {
-                                if let Err(e) = progress_tx.send(log_latest_block_number) {
-                                    error!("failed to send watch progress, error={:?}", e);
-                                } else {
-                                    continue;
-                                }
-                            }
-                        }
-                    }
-
-                    if stop {
+                LogFetchProgress::Transaction(tx) => {
+                    if !self.put_tx(tx.clone()).await {
                         // Unexpected error.
                         bail!("log sync write error");
                     }
@@ -455,7 +426,7 @@ impl LogSyncManager {
                 &executor_clone,
                 Duration::from_millis(self.config.recover_query_delay),
             );
-            self.handle_data(recover_rx, &None).await?;
+            self.handle_data(recover_rx).await?;
         }
 
         self.log_fetcher.start_remove_finalized_block_task(
@@ -469,33 +440,6 @@ impl LogSyncManager {
     }
 }
 
-async fn get_start_block_number_with_hash(
-    log_sync_manager: &LogSyncManager,
-) -> Result<(u64, H256), anyhow::Error> {
-    if let Some(block_number) = log_sync_manager.store.get_log_latest_block_number()? {
-        if let Some(Some(val)) = log_sync_manager
-            .block_hash_cache
-            .read()
-            .await
-            .get(&block_number)
-        {
-            return Ok((block_number, val.block_hash));
-        }
-    }
-
-    let (start_block_number, start_block_hash) = match log_sync_manager.store.get_sync_progress()? {
-        // No previous progress, so just use config.
-        None => {
-            let block_number = log_sync_manager.config.start_block_number;
-            let block_hash = log_sync_manager.get_block(block_number.into()).await?.1;
-            (block_number, block_hash)
-        }
-        Some((block_number, block_hash)) => (block_number, block_hash),
-    };
-
-    Ok((start_block_number, start_block_hash))
-}
-
 async fn run_and_log<R, E>(
     mut on_error: impl FnMut(),
     f: impl Future<Output = std::result::Result<R, E>> + Send,

node/storage/src/log_store/log_manager.rs

@@ -345,10 +345,6 @@ impl LogStoreWrite for LogManager {
         self.tx_store.put_progress(progress)
     }
 
-    fn put_log_latest_block_number(&self, block_number: u64) -> Result<()> {
-        self.tx_store.put_log_latest_block_number(block_number)
-    }
-
     /// Return the reverted Transactions in order.
     /// `tx_seq == u64::MAX` is a special case for reverting all transactions.
     fn revert_to(&self, tx_seq: u64) -> Result<Vec<Transaction>> {
@@ -538,10 +534,6 @@ impl LogStoreRead for LogManager {
         self.tx_store.get_progress()
     }
 
-    fn get_log_latest_block_number(&self) -> Result<Option<u64>> {
-        self.tx_store.get_log_latest_block_number()
-    }
-
     fn get_block_hash_by_number(&self, block_number: u64) -> Result<Option<(H256, Option<u64>)>> {
         self.tx_store.get_block_hash_by_number(block_number)
     }

node/storage/src/log_store/mod.rs

@@ -59,8 +59,6 @@ pub trait LogStoreRead: LogStoreChunkRead {
 
     fn get_sync_progress(&self) -> Result<Option<(u64, H256)>>;
 
-    fn get_log_latest_block_number(&self) -> Result<Option<u64>>;
-
     fn get_block_hash_by_number(&self, block_number: u64) -> Result<Option<(H256, Option<u64>)>>;
 
     fn get_block_hashes(&self) -> Result<Vec<(u64, BlockHashAndSubmissionIndex)>>;
@@ -128,9 +126,6 @@ pub trait LogStoreWrite: LogStoreChunkWrite {
     /// Store the progress of synced block number and its hash.
     fn put_sync_progress(&self, progress: (u64, H256, Option<Option<u64>>)) -> Result<()>;
 
-    /// Store the latest block number which has log
-    fn put_log_latest_block_number(&self, block_number: u64) -> Result<()>;
-
     /// Revert the log state to a given tx seq.
     /// This is needed when transactions are reverted because of chain reorg.
     ///

node/storage/src/log_store/tx_store.rs

@@ -19,7 +19,6 @@ use tracing::{error, instrument};
 
 const LOG_SYNC_PROGRESS_KEY: &str = "log_sync_progress";
 const NEXT_TX_KEY: &str = "next_tx_seq";
-const LOG_LATEST_BLOCK_NUMBER_KEY: &str = "log_latest_block_number_key";
 
 const TX_STATUS_FINALIZED: u8 = 0;
 const TX_STATUS_PRUNED: u8 = 1;
@@ -215,25 +214,6 @@ impl TransactionStore {
         ))
     }
 
-    #[instrument(skip(self))]
-    pub fn put_log_latest_block_number(&self, block_number: u64) -> Result<()> {
-        Ok(self.kvdb.put(
-            COL_MISC,
-            LOG_LATEST_BLOCK_NUMBER_KEY.as_bytes(),
-            &block_number.as_ssz_bytes(),
-        )?)
-    }
-
-    #[instrument(skip(self))]
-    pub fn get_log_latest_block_number(&self) -> Result<Option<u64>> {
-        Ok(Some(
-            <u64>::from_ssz_bytes(&try_option!(self
-                .kvdb
-                .get(COL_MISC, LOG_LATEST_BLOCK_NUMBER_KEY.as_bytes())?))
-            .map_err(Error::from)?,
-        ))
-    }
-
     pub fn get_block_hash_by_number(
         &self,
         block_number: u64,