[consensus] add missing checks

consensus/consensus-types/src/quorum_cert.rs

@@ -5,8 +5,8 @@ use crate::vote_data::VoteData;
 use anyhow::{ensure, Context};
 use aptos_bitvec::BitVec;
 use aptos_crypto::{hash::CryptoHash, HashValue};
-use aptos_types::aggregate_signature::AggregateSignature;
 use aptos_types::{
+    aggregate_signature::AggregateSignature,
     block_info::BlockInfo,
     ledger_info::{LedgerInfo, LedgerInfoWithSignatures},
     validator_verifier::ValidatorVerifier,

consensus/consensus-types/src/sync_info.rs

@@ -126,6 +126,10 @@ impl SyncInfo {
             epoch == self.highest_ordered_cert().certified_block().epoch(),
             "Multi epoch in SyncInfo - HOC and HQC"
         );
+        ensure!(
+            epoch == self.highest_commit_cert().commit_info().epoch(),
+            "Multi epoch in SyncInfo - HOC and HCC"
+        );
         if let Some(tc) = &self.highest_2chain_timeout_cert {
             ensure!(epoch == tc.epoch(), "Multi epoch in SyncInfo - TC and HQC");
         }

consensus/src/experimental/pipeline_phase.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::experimental::buffer_manager::{Receiver, Sender};
+use aptos_logger::debug;
 use async_trait::async_trait;
 use futures::{SinkExt, StreamExt};
 use std::sync::{
@@ -71,6 +72,7 @@ impl<T: StatelessPipeline> PipelinePhase<T> {
             let response = self.processor.process(req).await;
             if let Some(tx) = &mut self.maybe_tx {
                 if tx.send(response).await.is_err() {
+                    debug!("Failed to send response, buffer manager probably dropped");
                     break;
                 }
             }

consensus/src/round_manager.rs

@@ -815,7 +815,7 @@ impl RoundManager {
             VoteReceptionResult::VoteAdded(_)
             | VoteReceptionResult::EchoTimeout(_)
             | VoteReceptionResult::DuplicateVote => Ok(()),
-            e @ _ => Err(anyhow::anyhow!("{:?}", e)),
+            e => Err(anyhow::anyhow!("{:?}", e)),
         }
     }
 

consensus/src/txn_notifier.rs

@@ -66,7 +66,7 @@ impl TxnNotifier for MempoolNotifier {
         let compute_status = compute_results.compute_status();
         if txns.len() + 2 != compute_status.len() {
             // reconfiguration suffix blocks don't have any transactions
-            if compute_status.len() == 0 {
+            if compute_status.is_empty() {
                 return Ok(());
             }
             return Err(format_err!(