[Consensus Observer] Handle state sync notification race condition.

Author: JoshLind

consensus/src/consensus_observer/observer/block_data.rs

@@ -89,17 +89,6 @@ impl ObserverBlockData {
         self.block_payload_store.all_payloads_exist(blocks)
     }
 
-    /// Returns true iff the root epoch and round match the given values
-    pub fn check_root_epoch_and_round(&self, epoch: u64, round: Round) -> bool {
-        // Get the expected epoch and round
-        let root = self.root();
-        let expected_epoch = root.commit_info().epoch();
-        let expected_round = root.commit_info().round();
-
-        // Check if the expected epoch and round match the given values
-        expected_epoch == epoch && expected_round == round
-    }
-
     /// Clears all block data and returns the root ledger info
     pub fn clear_block_data(&mut self) -> LedgerInfoWithSignatures {
         // Clear the payload store
@@ -403,33 +392,6 @@ mod test {
         assert!(observer_block_data.all_payloads_exist(&[pipelined_block]));
     }
 
-    #[test]
-    fn test_check_root_epoch_and_round() {
-        // Create a root ledger info
-        let epoch = 10;
-        let round = 5;
-        let root = create_ledger_info(epoch, round);
-
-        // Create the observer block data
-        let mut observer_block_data =
-            ObserverBlockData::new_with_root(ConsensusObserverConfig::default(), root);
-
-        // Check the root epoch and round
-        assert!(observer_block_data.check_root_epoch_and_round(epoch, round));
-        assert!(!observer_block_data.check_root_epoch_and_round(epoch, round + 1));
-        assert!(!observer_block_data.check_root_epoch_and_round(epoch + 1, round));
-
-        // Update the root ledger info
-        let new_epoch = epoch + 10;
-        let new_round = round + 100;
-        let new_root = create_ledger_info(new_epoch, new_round);
-        observer_block_data.update_root(new_root.clone());
-
-        // Check the updated root epoch and round
-        assert!(!observer_block_data.check_root_epoch_and_round(epoch, round));
-        assert!(observer_block_data.check_root_epoch_and_round(new_epoch, new_round));
-    }
-
     #[test]
     fn test_clear_block_data() {
         // Create a root ledger info

consensus/src/consensus_observer/observer/consensus_observer.rs

@@ -972,14 +972,14 @@ impl ConsensusObserver {
     ) {
         // Get the epoch and round for the synced commit decision
         let ledger_info = latest_synced_ledger_info.ledger_info();
-        let epoch = ledger_info.epoch();
-        let round = ledger_info.round();
+        let synced_epoch = ledger_info.epoch();
+        let synced_round = ledger_info.round();
 
         // Log the state sync notification
         info!(
             LogSchema::new(LogEntry::ConsensusObserver).message(&format!(
-                "Received state sync notification for commit completion! Epoch {}, round: {}!",
-                epoch, round
+                "Received state sync notification for commit completion! Synced epoch {}, round: {}!",
+                synced_epoch, synced_round
             ))
         );
 
@@ -992,26 +992,41 @@ impl ConsensusObserver {
             return;
         }
 
-        // Verify that the state sync notification is for the current epoch and round
-        if !self
-            .observer_block_data
-            .lock()
-            .check_root_epoch_and_round(epoch, round)
-        {
+        // Get the block data root epoch and round
+        let block_data_root = self.observer_block_data.lock().root();
+        let block_data_epoch = block_data_root.ledger_info().epoch();
+        let block_data_round = block_data_root.ledger_info().round();
+
+        // If the commit sync notification is behind the block data root, ignore it. This
+        // is possible due to a race condition where we started syncing to a newer commit
+        // at the same time that state sync sent the notification for a previous commit.
+        if (synced_epoch, synced_round) < (block_data_epoch, block_data_round) {
+            info!(
+                LogSchema::new(LogEntry::ConsensusObserver).message(&format!(
+                    "Ignoring old commit sync notification for epoch: {}, round: {}! Current root: {:?}",
+                    synced_epoch, synced_round, block_data_root
+                ))
+            );
+            return;
+        }
+
+        // If the commit sync notification is ahead the block data root, something has gone wrong!
+        if (synced_epoch, synced_round) > (block_data_epoch, block_data_round) {
             // Log the error, reset the state sync manager and return early
             error!(
                 LogSchema::new(LogEntry::ConsensusObserver).message(&format!(
                     "Received invalid commit sync notification for epoch: {}, round: {}! Current root: {:?}",
-                    epoch, round, self.observer_block_data.lock().root()
+                    synced_epoch, synced_round, block_data_root
                 ))
             );
             self.state_sync_manager.clear_active_commit_sync();
             return;
         }
 
-        // If the epoch has changed, end the current epoch and start the latest one
+        // Otherwise, the commit sync notification matches the block data root.
+        // If the epoch has changed, end the current epoch and start the latest one.
         let current_epoch_state = self.get_epoch_state();
-        if epoch > current_epoch_state.epoch {
+        if synced_epoch > current_epoch_state.epoch {
             // Wait for the latest epoch to start
             self.execution_client.end_epoch().await;
             self.wait_for_epoch_start().await;