feat: compact block async

Author: driftluo

sync/src/relayer/block_transactions_process.rs

@@ -42,7 +42,7 @@ impl<'a> BlockTransactionsProcess<'a> {
         }
     }
 
-    pub fn execute(self) -> Status {
+    pub async fn execute(self) -> Status {
         let shared = self.relayer.shared();
         let active_chain = shared.active_chain();
         let block_transactions = self.message.to_entity();
@@ -65,6 +65,7 @@ impl<'a> BlockTransactionsProcess<'a> {
         if let Entry::Occupied(mut pending) = shared
             .state()
             .pending_compact_blocks()
+            .await
             .entry(block_hash.clone())
         {
             let (compact_block, peers_map, _) = pending.get_mut();
@@ -87,13 +88,16 @@ impl<'a> BlockTransactionsProcess<'a> {
                     &received_uncles,
                 ));
 
-                let ret = self.relayer.reconstruct_block(
-                    &active_chain,
-                    compact_block,
-                    received_transactions,
-                    expected_uncle_indexes,
-                    &received_uncles,
-                );
+                let ret = self
+                    .relayer
+                    .reconstruct_block(
+                        &active_chain,
+                        compact_block,
+                        received_transactions,
+                        expected_uncle_indexes,
+                        &received_uncles,
+                    )
+                    .await;
 
                 // Request proposal
                 {
@@ -167,13 +171,7 @@ impl<'a> BlockTransactionsProcess<'a> {
                     .build();
                 let message = packed::RelayMessage::new_builder().set(content).build();
 
-                self.relayer
-                    .shared()
-                    .shared()
-                    .async_handle()
-                    .spawn(
-                        async move { send_message_to_async(&self.nc, self.peer, &message).await },
-                    );
+                let _ignore = send_message_to_async(&self.nc, self.peer, &message).await;
 
                 let _ignore_prev_value =
                     mem::replace(expected_transaction_indexes, missing_transactions);

sync/src/relayer/compact_block_process.rs

@@ -53,7 +53,7 @@ impl<'a> CompactBlockProcess<'a> {
         }
     }
 
-    pub fn execute(self) -> Status {
+    pub async fn execute(self) -> Status {
         let instant = Instant::now();
         let shared = self.relayer.shared();
         let active_chain = shared.active_chain();
@@ -67,7 +67,7 @@ impl<'a> CompactBlockProcess<'a> {
             return status;
         }
 
-        let status = contextual_check(&header, shared, &active_chain, &self.nc, self.peer);
+        let status = contextual_check(&header, shared, &active_chain, &self.nc, self.peer).await;
         if !status.is_ok() {
             return status;
         }
@@ -89,7 +89,8 @@ impl<'a> CompactBlockProcess<'a> {
         // Reconstruct block
         let ret = self
             .relayer
-            .reconstruct_block(&active_chain, &compact_block, vec![], &[], &[]);
+            .reconstruct_block(&active_chain, &compact_block, vec![], &[], &[])
+            .await;
 
         // Accept block
         // `relayer.accept_block` will make sure the validity of block before persisting
@@ -102,7 +103,7 @@ impl<'a> CompactBlockProcess<'a> {
                         .inc_by(block.transactions().len() as u64);
                     metrics.ckb_relay_cb_reconstruct_ok.inc();
                 }
-                let mut pending_compact_blocks = shared.state().pending_compact_blocks();
+                let mut pending_compact_blocks = shared.state().pending_compact_blocks().await;
                 pending_compact_blocks.remove(&block_hash);
                 // remove all pending request below this block epoch
                 //
@@ -144,7 +145,8 @@ impl<'a> CompactBlockProcess<'a> {
                     missing_transactions,
                     missing_uncles,
                     self.peer,
-                );
+                )
+                .await;
 
                 StatusCode::CompactBlockRequiresFreshTransactions.with_context(&block_hash)
             }
@@ -163,7 +165,8 @@ impl<'a> CompactBlockProcess<'a> {
                     missing_transactions,
                     missing_uncles,
                     self.peer,
-                );
+                )
+                .await;
                 StatusCode::CompactBlockMeetsShortIdsCollision.with_context(&block_hash)
             }
             ReconstructionResult::Error(status) => status,
@@ -224,7 +227,7 @@ fn non_contextual_check(
 /// * check compact block's parent block is not stored in db
 /// * check compact block is in pending
 /// * check compact header verification
-fn contextual_check(
+async fn contextual_check(
     compact_block_header: &HeaderView,
     shared: &Arc<SyncShared>,
     active_chain: &ActiveChain,
@@ -278,7 +281,7 @@ fn contextual_check(
     }
 
     // compact block is in pending
-    let pending_compact_blocks = shared.state().pending_compact_blocks();
+    let pending_compact_blocks = shared.state().pending_compact_blocks().await;
     if pending_compact_blocks
         .get(&block_hash)
         .map(|(_, peers_map, _)| peers_map.contains_key(&peer))
@@ -339,7 +342,7 @@ fn contextual_check(
 }
 
 /// request missing txs and uncles from peer
-fn missing_or_collided_post_process(
+async fn missing_or_collided_post_process(
     compact_block: CompactBlock,
     block_hash: Byte32,
     shared: &SyncShared,
@@ -351,6 +354,7 @@ fn missing_or_collided_post_process(
     shared
         .state()
         .pending_compact_blocks()
+        .await
         .entry(block_hash.clone())
         .or_insert_with(|| (compact_block, HashMap::default(), unix_time_as_millis()))
         .1

sync/src/relayer/get_block_proposal_process.rs

@@ -53,7 +53,7 @@ impl<'a> GetBlockProposalProcess<'a> {
 
         let fetched_transactions = {
             let tx_pool = self.relayer.shared.shared().tx_pool_controller();
-            let fetch_txs = tx_pool.fetch_txs_async(proposals.clone()).await;
+            let fetch_txs = tx_pool.fetch_txs(proposals.clone()).await;
             if let Err(e) = fetch_txs {
                 debug_target!(
                     crate::LOG_TARGET_RELAY,

sync/src/relayer/mod.rs

@@ -121,7 +121,9 @@ impl Relayer {
 
         match message {
             packed::RelayMessageUnionReader::CompactBlock(reader) => {
-                CompactBlockProcess::new(reader, self, nc, peer).execute()
+                CompactBlockProcess::new(reader, self, nc, peer)
+                    .execute()
+                    .await
             }
             packed::RelayMessageUnionReader::RelayTransactions(reader) => {
                 if reader.check_data() {
@@ -146,7 +148,9 @@ impl Relayer {
             }
             packed::RelayMessageUnionReader::BlockTransactions(reader) => {
                 if reader.check_data() {
-                    BlockTransactionsProcess::new(reader, self, nc, peer).execute()
+                    BlockTransactionsProcess::new(reader, self, nc, peer)
+                        .execute()
+                        .await
                 } else {
                     StatusCode::ProtocolMessageIsMalformed
                         .with_context("BlockTransactions is invalid")
@@ -337,7 +341,7 @@ impl Relayer {
     // then once the block has been reconstructed, it shall be processed as normal,
     // keeping in mind that short_ids are expected to occasionally collide,
     // and that nodes must not be penalized for such collisions, wherever they appear.
-    pub fn reconstruct_block(
+    pub async fn reconstruct_block(
         &self,
         active_chain: &ActiveChain,
         compact_block: &packed::CompactBlock,
@@ -371,7 +375,7 @@ impl Relayer {
 
         if !short_ids_set.is_empty() {
             let tx_pool = self.shared.shared().tx_pool_controller();
-            let fetch_txs = tx_pool.fetch_txs(short_ids_set);
+            let fetch_txs = tx_pool.fetch_txs(short_ids_set).await;
             if let Err(e) = fetch_txs {
                 return ReconstructionResult::Error(StatusCode::TxPool.with_context(e));
             }
@@ -537,7 +541,7 @@ impl Relayer {
         let tx_pool = self.shared.shared().tx_pool_controller();
 
         let fetch_txs = tx_pool
-            .fetch_txs_async(
+            .fetch_txs(
                 get_block_proposals
                     .iter()
                     .map(|kv_pair| kv_pair.key().clone())

sync/src/relayer/tests/block_transactions_process.rs

@@ -21,6 +21,10 @@ fn test_accept_block() {
     let (_chain, relayer, _) = build_chain(5);
     let peer_index: PeerIndex = 100.into();
     let other_peer_index: PeerIndex = 101.into();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
 
     let tx1 = TransactionBuilder::default().build();
     let tx2 = TransactionBuilder::default()
@@ -54,7 +58,8 @@ fn test_accept_block() {
     let hash = compact_block.header().calc_header_hash();
 
     {
-        let mut pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let mut pending_compact_blocks =
+            rt.block_on(relayer.shared.state().pending_compact_blocks());
         pending_compact_blocks.insert(
             hash.clone(),
             (
@@ -84,9 +89,9 @@ fn test_accept_block() {
         peer_index,
     );
 
-    assert_eq!(process.execute(), Status::ok());
+    assert_eq!(rt.block_on(process.execute()), Status::ok());
 
-    let pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+    let pending_compact_blocks = rt.block_on(relayer.shared.state().pending_compact_blocks());
     assert!(pending_compact_blocks.get(&hash).is_none());
 
     std::thread::sleep(std::time::Duration::from_millis(100));
@@ -102,6 +107,10 @@ fn test_accept_block() {
 fn test_unknown_request() {
     let (_chain, relayer, _) = build_chain(5);
     let peer_index: PeerIndex = 100.into();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
 
     let tx1 = TransactionBuilder::default().build();
     let tx2 = TransactionBuilder::default()
@@ -123,7 +132,8 @@ fn test_unknown_request() {
 
     let foo_peer_index: PeerIndex = 998.into();
     {
-        let mut pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let mut pending_compact_blocks =
+            rt.block_on(relayer.shared.state().pending_compact_blocks());
         pending_compact_blocks.insert(
             compact_block.header().calc_header_hash(),
             (
@@ -148,13 +158,17 @@ fn test_unknown_request() {
         Arc::<MockProtocolContext>::clone(&nc),
         peer_index,
     );
-    assert_eq!(process.execute(), Status::ignored());
+    assert_eq!(rt.block_on(process.execute()), Status::ignored());
 }
 
 #[test]
 fn test_invalid_transaction_root() {
     let (_chain, relayer, _) = build_chain(5);
     let peer_index: PeerIndex = 100.into();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
 
     let tx1 = TransactionBuilder::default().build();
     let tx2 = TransactionBuilder::default()
@@ -188,7 +202,8 @@ fn test_invalid_transaction_root() {
     let block_hash = compact_block.header().calc_header_hash();
 
     {
-        let mut pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let mut pending_compact_blocks =
+            rt.block_on(relayer.shared.state().pending_compact_blocks());
         pending_compact_blocks.insert(
             block_hash.clone(),
             (
@@ -214,14 +229,18 @@ fn test_invalid_transaction_root() {
         peer_index,
     );
     assert_eq!(
-        process.execute(),
+        rt.block_on(process.execute()),
         StatusCode::CompactBlockHasUnmatchedTransactionRootWithReconstructedBlock.into(),
     );
 }
 
 #[test]
 fn test_collision_and_send_missing_indexes() {
     let (_chain, relayer, _) = build_chain(5);
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
 
     let active_chain = relayer.shared.active_chain();
     let last_block = relayer
@@ -284,7 +303,8 @@ fn test_collision_and_send_missing_indexes() {
 
     let hash = compact_block.header().calc_header_hash();
     {
-        let mut pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let mut pending_compact_blocks =
+            rt.block_on(relayer.shared.state().pending_compact_blocks());
         pending_compact_blocks.insert(
             hash.clone(),
             (
@@ -310,7 +330,7 @@ fn test_collision_and_send_missing_indexes() {
         peer_index,
     );
     assert_eq!(
-        process.execute(),
+        rt.block_on(process.execute()),
         StatusCode::CompactBlockMeetsShortIdsCollision.into()
     );
 
@@ -327,7 +347,7 @@ fn test_collision_and_send_missing_indexes() {
 
     // update cached missing_index
     {
-        let pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let pending_compact_blocks = rt.block_on(relayer.shared.state().pending_compact_blocks());
         assert_eq!(
             pending_compact_blocks
                 .get(&hash)
@@ -354,7 +374,7 @@ fn test_collision_and_send_missing_indexes() {
         peer_index,
     );
     assert_eq!(
-        process.execute(),
+        rt.block_on(process.execute()),
         StatusCode::CompactBlockHasUnmatchedTransactionRootWithReconstructedBlock.into(),
     );
 }
@@ -369,6 +389,10 @@ fn test_missing() {
 
     let (_chain, relayer, _) = build_chain(5);
     let peer_index: PeerIndex = 100.into();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
 
     let tx1 = TransactionBuilder::default().build();
     let tx2 = TransactionBuilder::default()
@@ -399,7 +423,8 @@ fn test_missing() {
     // tx3 should be in tx_pool already, but it's not.
     // so the reconstruct block will fail
     {
-        let mut pending_compact_blocks = relayer.shared.state().pending_compact_blocks();
+        let mut pending_compact_blocks =
+            rt.block_on(relayer.shared.state().pending_compact_blocks());
         pending_compact_blocks.insert(
             compact_block.header().calc_header_hash(),
             (
@@ -425,7 +450,7 @@ fn test_missing() {
         peer_index,
     );
     assert_eq!(
-        process.execute(),
+        rt.block_on(process.execute()),
         StatusCode::CompactBlockRequiresFreshTransactions.into()
     );