P2p: call preliminary_headers_check earlier

ImplOfAnImpl · ImplOfAnImpl · commit 89cbc5311a30 · 2026-03-31T10:52:17.000+03:00
diff --git a/p2p/src/sync/peer/block_manager.rs b/p2p/src/sync/peer/block_manager.rs
@@ -716,6 +716,15 @@ where
             self.incoming.peers_best_block_that_we_have.as_displayable()
         );
 
+        // Now use preliminary_headers_check; this can be done because the first header
+        // is known to be connected to the chainstate.
+        {
+            let new_block_headers = new_block_headers.clone();
+            self.chainstate_handle
+                .call(move |c| Ok(c.preliminary_headers_check(&new_block_headers)?))
+                .await?;
+        }
+
         if !self.incoming.requested_blocks.is_empty() {
             // We are already downloading blocks, so bail out.
             // Note that we unconditionally replace pending_headers with new_block_headers
@@ -732,15 +741,6 @@ where
             return Ok(());
         }
 
-        // Now use preliminary_headers_check; this can be done because the first header
-        // is known to be connected to the chainstate.
-        {
-            let new_block_headers = new_block_headers.clone();
-            self.chainstate_handle
-                .call(move |c| Ok(c.preliminary_headers_check(&new_block_headers)?))
-                .await?;
-        }
-
         self.request_blocks(new_block_headers)
     }
 
diff --git a/p2p/src/sync/tests/header_list_response.rs b/p2p/src/sync/tests/header_list_response.rs
@@ -15,16 +15,27 @@
 
 use std::{iter, sync::Arc, time::Duration};
 
+use itertools::Itertools as _;
+
 use chainstate::ban_score::BanScore;
 use chainstate_test_framework::TestFramework;
-use common::{chain::config::create_unit_test_config, primitives::Idable};
+use common::{
+    chain::{config::create_unit_test_config, GenBlock},
+    primitives::{BlockHeight, Id, Idable},
+};
+use logging::log;
 use p2p_test_utils::create_n_blocks;
-use test_utils::random::Seed;
+use randomness::Rng as _;
+use test_utils::{
+    assert_matches,
+    random::{make_seedable_rng, Seed},
+    BasicTestTimeGetter,
+};
 
 use crate::{
     error::ProtocolError,
-    message::{BlockListRequest, BlockSyncMessage, HeaderList},
-    sync::tests::helpers::TestNode,
+    message::{BlockListRequest, BlockResponse, BlockSyncMessage, HeaderList},
+    sync::{test_helpers::make_new_blocks, tests::helpers::TestNode},
     test_helpers::{for_each_protocol_version, test_p2p_config},
     types::peer_id::PeerId,
     P2pConfig, P2pError,
@@ -37,7 +48,7 @@ use crate::{
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn header_count_limit_exceeded(#[case] seed: Seed) {
     for_each_protocol_version(|protocol_version| async move {
-        let mut rng = test_utils::random::make_seedable_rng(seed);
+        let mut rng = make_seedable_rng(seed);
 
         let chain_config = Arc::new(create_unit_test_config());
         let mut tf = TestFramework::builder(&mut rng)
@@ -83,7 +94,7 @@ async fn header_count_limit_exceeded(#[case] seed: Seed) {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn unordered_headers(#[case] seed: Seed) {
     for_each_protocol_version(|protocol_version| async move {
-        let mut rng = test_utils::random::make_seedable_rng(seed);
+        let mut rng = make_seedable_rng(seed);
 
         let chain_config = Arc::new(create_unit_test_config());
         let mut tf = TestFramework::builder(&mut rng)
@@ -128,7 +139,7 @@ async fn unordered_headers(#[case] seed: Seed) {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn disconnected_headers(#[case] seed: Seed) {
     for_each_protocol_version(|protocol_version| async move {
-        let mut rng = test_utils::random::make_seedable_rng(seed);
+        let mut rng = make_seedable_rng(seed);
 
         let chain_config = Arc::new(create_unit_test_config());
         let mut tf = TestFramework::builder(&mut rng)
@@ -171,7 +182,7 @@ async fn disconnected_headers(#[case] seed: Seed) {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn valid_headers(#[case] seed: Seed) {
     for_each_protocol_version(|protocol_version| async move {
-        let mut rng = test_utils::random::make_seedable_rng(seed);
+        let mut rng = make_seedable_rng(seed);
 
         let chain_config = Arc::new(create_unit_test_config());
         let mut tf = TestFramework::builder(&mut rng)
@@ -246,3 +257,160 @@ async fn disconnect() {
     })
     .await;
 }
+
+// When handling a header list, the node must check headers' validity first, and only then check
+// if some blocks have already been requested.
+// The actual test happens with "make_branch2_invalid=true":
+// 1) Make a forked chain with 2 branches; the 1st one is good, the 2nd one has an invalid header.
+// 2) The peer sends a header list for the 1st branch; the node responds with a block list request.
+// 3) The peer sends a header list for the 2nd branch.
+// Expected result: the node sees that the new header list has an invalid header and adjusts
+// the peer score immediately.
+// The case "make_branch2_invalid=false" exists for completeness; all headers on the 2nd branch
+// are ok in this case and the node is expected to send another BlockListRequest and accept the
+// corresponding blocks without punishing the peer.
+#[tracing::instrument(skip(seed))]
+#[rstest::rstest]
+#[trace]
+#[case(Seed::from_entropy())]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn header_check_happens_before_checking_if_blocks_were_requested(
+    #[case] seed: Seed,
+    #[values(false, true)] make_branch2_invalid: bool,
+) {
+    for_each_protocol_version(|protocol_version| async move {
+        let mut rng = make_seedable_rng(seed);
+        let chain_config = Arc::new(common::chain::config::create_unit_test_config());
+        let time_getter = BasicTestTimeGetter::new();
+
+        let node_blocks_count = rng.gen_range(1..3);
+        let branch1_len = rng.gen_range(3..5);
+        let branch2_len = rng.gen_range(3..5);
+
+        log::debug!(
+            "node_blocks_count = {}, branch1_len = {}, branch2_len = {}",
+            node_blocks_count,
+            branch1_len,
+            branch2_len
+        );
+
+        let node_blocks = make_new_blocks(
+            &chain_config,
+            None,
+            &time_getter.get_time_getter(),
+            node_blocks_count,
+            &mut rng,
+        );
+
+        let branch1_blocks = make_new_blocks(
+            &chain_config,
+            node_blocks.last(),
+            &time_getter.get_time_getter(),
+            branch1_len,
+            &mut rng,
+        );
+
+        let branch2_blocks = make_new_blocks(
+            &chain_config,
+            node_blocks.last(),
+            &time_getter.get_time_getter(),
+            branch2_len,
+            &mut rng,
+        );
+
+        let chain_config = if make_branch2_invalid {
+            let branch2_invalid_header_idx =
+                rng.gen_range(0..std::cmp::min(branch1_blocks.len(), branch2_blocks.len()));
+
+            // To make one of the headers on branch2 invalid, specify a checkpoint referring to
+            // a block on branch1.
+            let checkpoint_height =
+                BlockHeight::new((branch2_invalid_header_idx + node_blocks.len() + 1) as u64);
+            let checkpoint_block_id: Id<GenBlock> =
+                branch1_blocks[branch2_invalid_header_idx].get_id().into();
+
+            let chain_config = Arc::new(
+                common::chain::config::create_unit_test_config_builder()
+                    .checkpoints([(checkpoint_height, checkpoint_block_id)].into())
+                    .build(),
+            );
+
+            chain_config
+        } else {
+            chain_config
+        };
+
+        let p2p_config = Arc::new(test_p2p_config());
+
+        let mut node = TestNode::builder(protocol_version)
+            .with_chain_config(chain_config)
+            .with_p2p_config(Arc::clone(&p2p_config))
+            .with_time_getter(time_getter.get_time_getter())
+            .with_blocks(node_blocks.clone())
+            .build()
+            .await;
+
+        let peer = node.connect_peer(PeerId::new(), protocol_version).await;
+
+        log::debug!("Peer sends HeaderList for branch1");
+        peer.send_headers(branch1_blocks.iter().map(|block| block.header().clone()).collect_vec())
+            .await;
+
+        log::debug!("Expecting BlockListRequest for branch 1");
+        assert_eq!(
+            node.get_sent_block_sync_message().await.1,
+            BlockSyncMessage::BlockListRequest(BlockListRequest::new(
+                branch1_blocks.iter().map(|block| block.get_id()).collect_vec()
+            )),
+        );
+
+        log::debug!("Peer sends HeaderList for branch2");
+        peer.send_headers(branch2_blocks.iter().map(|block| block.header().clone()).collect_vec())
+            .await;
+
+        if make_branch2_invalid {
+            // The node should check the headers right away and discover that one of them is invalid.
+            node.assert_peer_score_adjustment(peer.get_id(), 100).await;
+            node.assert_no_sync_message().await;
+        } else {
+            // If the headers are ok, the node should accept the previously requested blocks
+            // and make a new block request for the new ones.
+
+            for block in &branch1_blocks {
+                log::debug!("Peer sends BlockResponse for branch 1");
+                peer.send_block_sync_message(BlockSyncMessage::BlockResponse(BlockResponse::new(
+                    block.clone(),
+                )))
+                .await;
+            }
+
+            log::debug!("Expecting BlockListRequest for branch 2");
+            assert_eq!(
+                node.get_sent_block_sync_message().await.1,
+                BlockSyncMessage::BlockListRequest(BlockListRequest::new(
+                    branch2_blocks.iter().map(|block| block.get_id()).collect_vec()
+                )),
+            );
+
+            for block in &branch2_blocks {
+                log::debug!("Peer sends BlockResponse for branch 2");
+                peer.send_block_sync_message(BlockSyncMessage::BlockResponse(BlockResponse::new(
+                    block.clone(),
+                )))
+                .await;
+            }
+
+            log::debug!("Expecting final HeaderListRequest");
+            assert_matches!(
+                node.get_sent_block_sync_message().await.1,
+                BlockSyncMessage::HeaderListRequest(_)
+            );
+        }
+
+        node.assert_no_error().await;
+        node.assert_no_peer_manager_event().await;
+        node.assert_no_sync_message().await;
+        node.join_subsystem_manager().await;
+    })
+    .await;
+}
diff --git a/p2p/src/sync/tests/helpers/mod.rs b/p2p/src/sync/tests/helpers/mod.rs
@@ -251,6 +251,7 @@ impl TestNode {
 
     /// Panics if the sync manager returns an error.
     pub async fn assert_no_error(&mut self) {
+        log::debug!("Asserting no error");
         expect_no_recv!(self.error_receiver);
     }
 
@@ -362,6 +363,8 @@ impl TestNode {
     /// like NewTipReceived/NewChainstateTip etc).
     // TODO: Rename the function
     pub async fn assert_no_peer_manager_event(&mut self) {
+        log::debug!("Asserting no peer mgr event");
+
         time::timeout(SHORT_TIMEOUT, async {
             loop {
                 let peer_event = self.peer_manager_event_receiver.recv().await.unwrap();
@@ -400,10 +403,16 @@ impl TestNode {
 
     /// Panics if the sync manager sends a message.
     pub async fn assert_no_sync_message(&mut self) {
+        log::debug!("Asserting no sync message");
+
         let future = async {
             tokio::select! {
-                _ = self.block_sync_msg_receiver.recv() => {},
-                _ = self.transaction_sync_msg_receiver.recv() => {},
+                msg = self.block_sync_msg_receiver.recv() => {
+                    log::debug!("Got sync msg {msg:?} while expecting none");
+                },
+                msg = self.transaction_sync_msg_receiver.recv() => {
+                    log::debug!("Got sync msg {msg:?} while expecting none");
+                },
             }
         };
 
@@ -415,6 +424,8 @@ impl TestNode {
         expected_peer: PeerId,
         expected_score: u32,
     ) {
+        log::debug!("Asserting peer score adjustment");
+
         let (adjusted_peer, score) = self.receive_adjust_peer_score_event().await;
         assert_eq!(adjusted_peer, expected_peer);
         assert_eq!(score, expected_score);
@@ -429,6 +440,8 @@ impl TestNode {
     }
 
     pub async fn join_subsystem_manager(self) {
+        log::debug!("Joining subsystem mgr");
+
         // Shutdown sync manager first
         drop(self.syncing_event_sender);
         let _ = self.sync_manager_handle.await;
