Merge pull request #4562 from driftluo/fix-sync-relayer-collaboration

fix: fix sync relayer collaboration

sync/src/relayer/mod.rs

@@ -8,13 +8,13 @@ mod get_block_proposal_process;
 mod get_block_transactions_process;
 mod get_transactions_process;
 #[cfg(test)]
-mod tests;
+pub(crate) mod tests;
 mod transaction_hashes_process;
 mod transactions_process;
 
 use self::block_proposal_process::BlockProposalProcess;
 use self::block_transactions_process::BlockTransactionsProcess;
-use self::compact_block_process::CompactBlockProcess;
+pub(crate) use self::compact_block_process::CompactBlockProcess;
 use self::get_block_proposal_process::GetBlockProposalProcess;
 use self::get_block_transactions_process::GetBlockTransactionsProcess;
 use self::get_transactions_process::GetTransactionsProcess;

sync/src/relayer/tests/mod.rs

@@ -6,5 +6,5 @@ mod compact_block_process;
 mod compact_block_verifier;
 mod get_block_proposal_process;
 mod get_transactions_process;
-mod helper;
+pub(crate) mod helper;
 mod reconstruct_block;

sync/src/tests/sync_shared.rs

@@ -1,12 +1,15 @@
 use crate::block_status::BlockStatus;
+use crate::relayer::tests::helper::MockProtocolContext;
+use crate::relayer::CompactBlockProcess;
+use crate::synchronizer::HeadersProcess;
 use crate::tests::util::{build_chain, inherit_block};
-use crate::SyncShared;
+use crate::{Relayer, Status, SyncShared, Synchronizer};
 use ckb_chain::chain::ChainService;
 use ckb_shared::SharedBuilder;
 use ckb_store::{self, ChainStore};
 use ckb_test_chain_utils::always_success_cellbase;
 use ckb_types::core::Capacity;
-use ckb_types::prelude::*;
+use ckb_types::{packed, prelude::*};
 use std::sync::Arc;
 
 #[test]
@@ -183,3 +186,148 @@ fn test_switch_valid_fork() {
         );
     }
 }
+
+#[test]
+fn test_sync_relay_collaboration() {
+    let (shared, chain) = build_chain(2);
+    let sync_shared = Arc::new(shared);
+
+    let sync = Synchronizer::new(chain.clone(), Arc::clone(&sync_shared));
+    let relay = Relayer::new(chain, Arc::clone(&sync_shared));
+
+    let mock_relay_protocol_context =
+        MockProtocolContext::new(ckb_network::SupportProtocols::RelayV2);
+    let mock_sync_protocol_context = MockProtocolContext::new(ckb_network::SupportProtocols::Sync);
+
+    let relay_nc = Arc::new(mock_relay_protocol_context);
+    let sync_nc = Arc::new(mock_sync_protocol_context);
+
+    let new_block = {
+        let tip_hash = sync_shared.active_chain().tip_header().hash();
+        let next_block = inherit_block(sync_shared.shared(), &tip_hash).build();
+        Arc::new(next_block)
+    };
+
+    let compact_block_content =
+        packed::CompactBlock::build_from_block(&new_block, &std::collections::HashSet::new());
+
+    let headers_content = packed::SendHeaders::new_builder()
+        .headers([new_block.header()].map(|x| x.data()).pack())
+        .build();
+
+    // keep header process snapshot on old state, this is the bug reason
+    let header_process = HeadersProcess::new(
+        headers_content.as_reader(),
+        &sync,
+        1.into(),
+        sync_nc.as_ref(),
+    );
+
+    let compact_block_process = CompactBlockProcess::new(
+        compact_block_content.as_reader(),
+        &relay,
+        relay_nc as _,
+        1.into(),
+    );
+
+    let status = compact_block_process.execute();
+
+    assert!(status.is_ok());
+    assert_eq!(sync_shared.active_chain().tip_number(), new_block.number());
+
+    let status = header_process.execute();
+    assert!(status.is_ok());
+
+    assert_eq!(
+        sync_shared
+            .active_chain()
+            .get_block_status(&new_block.hash()),
+        BlockStatus::BLOCK_VALID
+    )
+}
+
+#[test]
+fn test_sync_relay_collaboration2() {
+    let (shared, chain) = build_chain(2);
+    let sync_shared = Arc::new(shared);
+
+    let sync = Synchronizer::new(chain.clone(), Arc::clone(&sync_shared));
+    let relay = Relayer::new(chain, Arc::clone(&sync_shared));
+
+    let mock_relay_protocol_context =
+        MockProtocolContext::new(ckb_network::SupportProtocols::RelayV2);
+    let mock_sync_protocol_context = MockProtocolContext::new(ckb_network::SupportProtocols::Sync);
+
+    let relay_nc = Arc::new(mock_relay_protocol_context);
+    let sync_nc = Arc::new(mock_sync_protocol_context);
+
+    let new_block = {
+        let tip_hash = sync_shared.active_chain().tip_header().hash();
+        let next_block = inherit_block(sync_shared.shared(), &tip_hash).build();
+        Arc::new(next_block)
+    };
+
+    let new_block_1 = {
+        let tip_hash = sync_shared.active_chain().tip_header().hash();
+        let next_block = inherit_block(sync_shared.shared(), &tip_hash).build();
+        let next_timestamp = next_block.timestamp() + 2;
+        let new_block = new_block
+            .as_advanced_builder()
+            .timestamp(next_timestamp.pack())
+            .build();
+
+        Arc::new(new_block)
+    };
+
+    let compact_block_content =
+        packed::CompactBlock::build_from_block(&new_block, &std::collections::HashSet::new());
+
+    let compact_block_content_1 =
+        packed::CompactBlock::build_from_block(&new_block_1, &std::collections::HashSet::new());
+
+    let headers_content = packed::SendHeaders::new_builder()
+        .headers([new_block.header()].map(|x| x.data()).pack())
+        .build();
+
+    // keep header process snapshot on old state, this is the bug reason
+    let header_process = HeadersProcess::new(
+        headers_content.as_reader(),
+        &sync,
+        1.into(),
+        sync_nc.as_ref(),
+    );
+
+    let compact_block_process = CompactBlockProcess::new(
+        compact_block_content.as_reader(),
+        &relay,
+        Arc::clone(&relay_nc) as _,
+        1.into(),
+    );
+
+    let status = compact_block_process.execute();
+
+    assert!(status.is_ok());
+    assert_eq!(sync_shared.active_chain().tip_number(), new_block.number());
+
+    let compact_block_process = CompactBlockProcess::new(
+        compact_block_content_1.as_reader(),
+        &relay,
+        relay_nc as _,
+        1.into(),
+    );
+
+    let status = compact_block_process.execute();
+
+    assert_eq!(status, Status::ok());
+    assert_eq!(sync_shared.active_chain().tip_number(), new_block.number());
+
+    let status = header_process.execute();
+    assert!(status.is_ok());
+
+    assert_eq!(
+        sync_shared
+            .active_chain()
+            .get_block_status(&new_block.hash()),
+        BlockStatus::BLOCK_VALID
+    )
+}

sync/src/types/mod.rs

@@ -1735,14 +1735,15 @@ impl SyncState {
     }
 
     pub fn may_set_shared_best_header(&self, header: HeaderIndexView) {
-        if !header.is_better_than(self.shared_best_header.read().total_difficulty()) {
+        let mut shared_best_header = self.shared_best_header.write();
+        if !header.is_better_than(shared_best_header.total_difficulty()) {
             return;
         }
 
         if let Some(metrics) = ckb_metrics::handle() {
             metrics.ckb_shared_best_number.set(header.number() as i64);
         }
-        *self.shared_best_header.write() = header;
+        *shared_best_header = header;
     }
 
     pub fn remove_header_view(&self, hash: &Byte32) {
@@ -2320,7 +2321,9 @@ impl ActiveChain {
                     BlockStatus::HEADER_VALID
                 } else {
                     let verified = self
-                        .snapshot
+                        .shared()
+                        .shared
+                        .snapshot()
                         .get_block_ext(block_hash)
                         .map(|block_ext| block_ext.verified);
                     match verified {