fix(consensus): Improve error handling for quorum cert retrieval (#443)

## PR Description
[Briefly describe your changes]
Issue Number: closes #xxx

## Tested?

- [ ] Yes
- [ ] No

Author: Lchangliang

aptos-core/consensus/consensus-types/src/block_retrieval.rs

@@ -104,6 +104,8 @@ pub enum BlockRetrievalStatus {
     NotEnoughBlocks,
     // Successfully found the target,
     SucceededWithTarget,
+    // Can not find the quorum certificate for the block.
+    QuorumCertNotFound,
 }
 
 /// Carries the returned blocks and the retrieval status.

aptos-core/consensus/src/block_storage/block_store.rs

@@ -223,6 +223,7 @@ impl BlockStore {
                 info!("sending qc {} to execution, current commit round {}", qc.commit_info().round(), self.commit_root().round());
                 if let Err(e) = self.send_for_execution(qc.into_wrapped_ledger_info(), true).await {
                     error!("Error in try-committing blocks. {}", e.to_string());
+                    break;
                 }
             }
         }

aptos-core/consensus/src/block_storage/sync_manager.rs

@@ -397,7 +397,7 @@ impl BlockStore {
         self.rebuild(root, blocks, quorum_certs).await;
         storage.consensus_db().ledger_db.metadata_db().set_latest_ledger_info(ledger_infos.last().unwrap().clone());
 
-        if ledger_infos.last().unwrap().ledger_info().ends_epoch() {
+        if !ledger_infos.is_empty() && ledger_infos.last().unwrap().ledger_info().ends_epoch() {
             retriever
                 .network
                 .send_epoch_change(EpochChangeProof::new(
@@ -625,7 +625,14 @@ impl BlockStore {
             let mut parent_id = HashValue::zero();
             let mut is_last_block = false;
             if let Some(executed_block) = self.get_block(id) {
-                let qc = self.get_quorum_cert_for_block(id).unwrap();
+                let qc = match self.get_quorum_cert_for_block(id) {
+                    Some(qc) => qc,
+                    None => {
+                        info!("Cannot find quorum cert for block id {}", id);
+                        status = BlockRetrievalStatus::QuorumCertNotFound;
+                        break;
+                    }
+                };
                 is_last_block = qc.commit_info().round() == 0;
                 quorum_certs.push((*qc).clone());  
                 let randomness = match executed_block.block().block_number() {
@@ -637,7 +644,19 @@ impl BlockStore {
             } else if let Ok(Some(executed_block)) =
                 self.storage.consensus_db().get_block(retrieval_epoch, id)
             {
-                let qc = self.storage.consensus_db().get_qc(retrieval_epoch, id).unwrap().unwrap();
+                let qc = match self.storage.consensus_db().get_qc(retrieval_epoch, id) {
+                    Ok(Some(qc)) => qc,
+                    Ok(None) => {
+                        info!("Cannot find quorum cert for block id {} in epoch {}", id, retrieval_epoch);
+                        status = BlockRetrievalStatus::QuorumCertNotFound;
+                        break;
+                    }
+                    Err(e) => {
+                        error!("Error retrieving quorum cert for block id {} in epoch {}: {:?}", id, retrieval_epoch, e);
+                        status = BlockRetrievalStatus::QuorumCertNotFound;
+                        break;
+                    }
+                };
                 is_last_block = qc.commit_info().round() == 0;
                 quorum_certs.push(qc);
                 let randomness = self.storage.consensus_db().get_randomness(executed_block.block_number().unwrap()).unwrap();

aptos-core/consensus/src/consensusdb/mod.rs

@@ -129,7 +129,8 @@ impl ConsensusDB {
         &self,
         latest_block_number: u64,
         epoch: u64,
-    ) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>, Vec<Block>, Vec<QuorumCert>)> {
+    ) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>, Vec<Block>, Vec<QuorumCert>, bool)> {
+        let mut has_root = false;
         let last_vote = self.get_last_vote()?;
         let highest_2chain_timeout_certificate = self.get_highest_2chain_timeout_certificate()?;
         let start_key = (epoch, HashValue::zero());
@@ -143,6 +144,7 @@ impl ConsensusDB {
         let (start_epoch, start_round, start_block_id) = if block_number_to_block_id.contains_key(&latest_block_number) {
             let block = self.get::<BlockSchema>(&(epoch, block_number_to_block_id[&latest_block_number]))?
                 .unwrap();
+            has_root = true;
             (block.epoch(), block.round(), block.id())
         } else {
             (epoch, 0, HashValue::zero())
@@ -170,7 +172,7 @@ impl ConsensusDB {
             .collect();
         info!("consensus_blocks size : {}, consensus_qcs size : {}, block_number_to_block_id size : {}, start_round : {}",
                  consensus_blocks.len(), consensus_qcs.len(), block_number_to_block_id.len(), start_round);
-        Ok((last_vote, highest_2chain_timeout_certificate, consensus_blocks, consensus_qcs))
+        Ok((last_vote, highest_2chain_timeout_certificate, consensus_blocks, consensus_qcs, has_root))
     }
 
     pub fn save_highest_2chain_timeout_certificate(&self, tc: Vec<u8>) -> Result<(), DbError> {

aptos-core/consensus/src/epoch_manager.rs

@@ -1597,7 +1597,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             | ConsensusMsg::SignedBatchInfo(_)
             | ConsensusMsg::ProofOfStoreMsg(_) => {
                 let event: UnverifiedEvent = msg.into();
-                info!("event epoch: {:?} and self epoch {:?}", event.epoch(), self.epoch());
+                debug!("event epoch: {:?} and self epoch {:?}", event.epoch(), self.epoch());
                 if event.epoch()? == self.epoch() {
                     return Ok(Some(event));
                 } else {

aptos-core/consensus/src/persistent_liveness_storage.rs

@@ -276,12 +276,13 @@ impl RecoveryData {
         mut quorum_certs: Vec<QuorumCert>,
         highest_2chain_timeout_cert: Option<TwoChainTimeoutCertificate>,
         order_vote_enabled: bool,
+        has_root: bool,
     ) -> Result<Self> {
         info!("blocks in db: {:?}", blocks.len());
         info!("quorum certs in db: {:?}", quorum_certs.len());
 
         let root;
-        if !blocks.is_empty() && execution_latest_block_num != ledger_recovery_data.storage_ledger.ledger_info().block_number() {
+        if has_root {
             root = Self::find_root_by_block_number(
                 execution_latest_block_num,
                 &mut blocks,
@@ -463,6 +464,7 @@ impl PersistentLivenessStorage for StorageWriteProxy {
             // Use original method
             self.aptos_db.get_latest_ledger_info().unwrap()
         };
+        info!("is_last_block_of_prev_epoch: {}, latest_ledger_info: {:?}, has_root: {}", is_last_block_of_prev_epoch, latest_ledger_info, raw_data.4);
 
         let ledger_recovery_data = LedgerRecoveryData::new(latest_ledger_info);
         match RecoveryData::new(
@@ -473,6 +475,7 @@ impl PersistentLivenessStorage for StorageWriteProxy {
             quorum_certs,
             highest_2chain_timeout_cert,
             order_vote_enabled,
+            raw_data.4,
         ) {
             Ok(initial_data) => {
                 // TODO(gravity_lightman)

aptos-core/consensus/src/util/db_tool.rs

@@ -41,7 +41,7 @@ impl Command {
         let all_batches = quorum_store_db.get_all_batches().unwrap();
 
         let consensus_db = ConsensusDB::new(self.db_dir.clone(), &PathBuf::new());
-        let (_, _, blocks, _) = consensus_db.get_data(0, 0)?;
+        let (_, _, blocks, _, _) = consensus_db.get_data(0, 0)?;
 
         let mut txns = Vec::new();
         for block in blocks {