[qs] tolerate block timestamp being updated asynchronously

In previous commit, we move timestamp update to async callback to ensure the consistency between mempool, qs and consensus.
However it introduces the race of timestamp can be updated by sync_to first then async callback and violates the assertion.

(cherry picked from commit 41e82f7)

consensus/src/quorum_store/batch_generator.rs

@@ -495,10 +495,10 @@ impl BatchGenerator {
                                 "QS: got clean request from execution, block timestamp {}",
                                 block_timestamp
                             );
-                            assert!(
-                                self.latest_block_timestamp <= block_timestamp,
-                                "Decreasing block timestamp"
-                            );
+                            // Block timestamp is updated asynchronously, so it may race when it enters state sync.
+                            if self.latest_block_timestamp > block_timestamp {
+                                continue;
+                            }
                             self.latest_block_timestamp = block_timestamp;
 
                             for (author, batch_id) in batches.iter().map(|b| (b.author(), b.batch_id())) {

consensus/src/quorum_store/batch_store.rs

@@ -332,17 +332,8 @@ impl BatchStore {
 
     pub fn update_certified_timestamp(&self, certified_time: u64) {
         trace!("QS: batch reader updating time {:?}", certified_time);
-        let prev_time = self
-            .last_certified_time
+        self.last_certified_time
             .fetch_max(certified_time, Ordering::SeqCst);
-        // Note: prev_time may be equal to certified_time due to state-sync
-        // at the epoch boundary.
-        assert!(
-            prev_time <= certified_time,
-            "Decreasing executed block timestamp reported to BatchReader {} {}",
-            prev_time,
-            certified_time,
-        );
 
         let expired_keys = self.clear_expired_payload(certified_time);
         if let Err(e) = self.db.delete_batches(expired_keys) {