[consensus] Additional round manager test and fixing block retrieval …

…panic (aptos-labs#5309)

consensus/src/block_storage/block_store_test.rs

@@ -119,14 +119,14 @@ proptest! {
             // match the signer_strategy in validator_signer.rs
             |key| Author::from_bytes(&key.public_key().to_bytes()[0..32]).unwrap()
         ).collect();
-        let mut runtime = consensus_runtime();
+        let runtime = consensus_runtime();
         let block_store = build_empty_tree();
         for block in blocks {
             if block.round() > 0 && authors.contains(&block.author().unwrap()) {
                 let known_parent = block_store.block_exists(block.parent_id());
                 let certified_parent = block.quorum_cert().certified_block().id() == block.parent_id();
                 let verify_res = block.verify_well_formed();
-                let res = timed_block_on(&mut runtime, block_store.execute_and_insert_block(block.clone()));
+                let res = timed_block_on(&runtime, block_store.execute_and_insert_block(block.clone()));
                 if !certified_parent {
                     prop_assert!(verify_res.is_err());
                 } else if !known_parent {

consensus/src/block_storage/sync_manager.rs

@@ -260,11 +260,15 @@ impl BlockStore {
             .iter()
             .any(|block| block.id() == highest_commit_cert.certified_block().id())
         {
+            info!(
+                "Found forked QC {}, fetching it as well",
+                highest_commit_cert
+            );
             let mut additional_blocks = retriever
                 .retrieve_block_for_qc(
                     highest_commit_cert,
                     1,
-                    highest_commit_cert.commit_info().id(),
+                    highest_commit_cert.certified_block().id(),
                 )
                 .await?;
 

consensus/src/experimental/tests/buffer_manager_tests.rs

@@ -237,7 +237,7 @@ fn buffer_manager_happy_path_test() {
         msg_tx,
         mut self_loop_rx,
         _hash_val,
-        mut runtime,
+        runtime,
         signers,
         mut result_rx,
         verifier,
@@ -268,7 +268,7 @@ fn buffer_manager_happy_path_test() {
         last_proposal = Some(proposal.last().unwrap().clone());
     }
 
-    timed_block_on(&mut runtime, async move {
+    timed_block_on(&runtime, async move {
         for i in 0..num_batches {
             block_tx
                 .send(OrderedBlocks {
@@ -299,7 +299,7 @@ fn buffer_manager_sync_test() {
         msg_tx,
         mut self_loop_rx,
         _hash_val,
-        mut runtime,
+        runtime,
         signers,
         mut result_rx,
         verifier,
@@ -332,7 +332,7 @@ fn buffer_manager_sync_test() {
 
     let dropped_batches = 42;
 
-    timed_block_on(&mut runtime, async move {
+    timed_block_on(&runtime, async move {
         for i in 0..dropped_batches {
             block_tx
                 .send(OrderedBlocks {

consensus/src/experimental/tests/phase_tester.rs

@@ -48,9 +48,9 @@ impl<T: StatelessPipeline> PhaseTester<T> {
     // unit tests are for phase processors only,
     // this function consumes the tester
     pub fn unit_test(self, processor: &T) {
-        let mut runtime = consensus_runtime();
+        let runtime = consensus_runtime();
 
-        timed_block_on(&mut runtime, async move {
+        timed_block_on(&runtime, async move {
             for PhaseTestCase {
                 index,
                 input,
@@ -76,9 +76,9 @@ impl<T: StatelessPipeline> PhaseTester<T> {
         mut tx: Sender<CountedRequest<T::Request>>,
         mut rx: Receiver<T::Response>,
     ) {
-        let mut runtime = consensus_runtime();
+        let runtime = consensus_runtime();
 
-        timed_block_on(&mut runtime, async move {
+        timed_block_on(&runtime, async move {
             for PhaseTestCase {
                 index,
                 input,

consensus/src/network_tests.rs

@@ -555,7 +555,7 @@ mod tests {
 
     #[test]
     fn test_network_api() {
-        let mut runtime = consensus_runtime();
+        let runtime = consensus_runtime();
         let num_nodes = 5;
         let mut receivers: Vec<NetworkReceivers> = Vec::new();
         let mut playground = NetworkPlayground::new(runtime.handle().clone());
@@ -628,7 +628,7 @@ mod tests {
             .unwrap(),
             SyncInfo::new(previous_qc.clone(), previous_qc, None),
         );
-        timed_block_on(&mut runtime, async {
+        timed_block_on(&runtime, async {
             nodes[0]
                 .send_vote(vote_msg.clone(), peers[2..5].to_vec())
                 .await;
@@ -658,7 +658,7 @@ mod tests {
 
     #[test]
     fn test_rpc() {
-        let mut runtime = consensus_runtime();
+        let runtime = consensus_runtime();
         let num_nodes = 2;
         let mut senders = Vec::new();
         let mut receivers: Vec<NetworkReceivers> = Vec::new();
@@ -746,7 +746,7 @@ mod tests {
         };
         runtime.handle().spawn(on_request_block);
         let peer = peers[1];
-        timed_block_on(&mut runtime, async {
+        timed_block_on(&runtime, async {
             let response = nodes[0]
                 .request_block(
                     BlockRetrievalRequest::new(HashValue::zero(), 1),
@@ -817,7 +817,7 @@ mod tests {
         };
         let f_network_task = network_task.start();
 
-        let mut runtime = consensus_runtime();
-        timed_block_on(&mut runtime, future::join(f_network_task, f_check));
+        let runtime = consensus_runtime();
+        timed_block_on(&runtime, future::join(f_network_task, f_check));
     }
 }

consensus/src/quorum_store/direct_mempool_quorum_store.rs

@@ -83,8 +83,8 @@ impl DirectMempoolQuorumStore {
         let (txns, result) = match payload_filter {
             PayloadFilter::DirectMempool(exclude_txns) => {
                 match self.pull_internal(max_txns, max_bytes, exclude_txns).await {
-                    Err(_) => {
-                        error!("GetBatch failed");
+                    Err(e) => {
+                        error!("GetBatch failed {:?}", e);
                         (vec![], counters::REQUEST_FAIL_LABEL)
                     }
                     Ok(txns) => (txns, counters::REQUEST_SUCCESS_LABEL),