Repository: mintlayer/mintlayer-core
Pull request #769 (2023-03-17 08:10:24)

Try to fix the problem with stuck sync
Functional tests started to fail after some recent changes.
Here is one such failed run: https://github.com/mintlayer/mintlayer-core/actions/runs/4442342875
The logs show this error:
```
Adjusting the '2' peer score by 20: ProtocolError(UnexpectedMessage("Peer requested already known block"))
```

It looks like the test fails this way:
1. Two new blocks submitted on Node0.
2. Node1 connects and downloads both blocks in initial block download.
3. Node0 sends new block announcement to all connected peers (this may happen with some delay because block announcements are sent from a different even loop).
4. Node1 receives this announcement and tries to download this already known block.
5. Node0 stops syncing because of the "Peer requested already known block" error.

To fix this, I added a check to drop block announcements for already known blocks.


Patch:
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index 5795efe2b..9e19a85cf 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -14,7 +14,10 @@
 // limitations under the License.
 
 use chainstate::ban_score::BanScore;
-use common::primitives::semver::SemVer;
+use common::{
+    chain::Block,
+    primitives::{semver::SemVer, Id},
+};
 use thiserror::Error;
 
 /// Errors related to invalid data/peer information that results in connection getting closed
@@ -25,10 +28,6 @@ pub enum ProtocolError {
     DifferentNetwork([u8; 4], [u8; 4]),
     #[error("Peer has an unsupported version. Our version {0}, their version {1}")]
     InvalidVersion(SemVer, SemVer),
-    // TODO: This error is very generic and probably should be replaced with several different ones,
-    // because it has a ban score of 100 and in many cases it is too harsh.
-    #[error("Peer sent an invalid message")]
-    InvalidMessage,
     #[error("Peer is unresponsive")]
     Unresponsive,
     #[error("Locator size ({0}) exceeds allowed limit ({1})")]
@@ -37,12 +36,20 @@ pub enum ProtocolError {
     BlocksRequestLimitExceeded(usize, usize),
     #[error("Number of headers in message ({0}) exceeds allowed limit ({1})")]
     HeadersLimitExceeded(usize, usize),
-    #[error("A peer requested an unknown block")]
-    UnknownBlockRequested,
+    #[error("A peer requested an unknown block ({0})")]
+    UnknownBlockRequested(Id<Block>),
+    #[error("A peer tried to download same block ({0})")]
+    DuplicatedBlockRequest(Id<Block>),
     #[error("Headers aren't connected")]
     DisconnectedHeaders,
     #[error("Received a message ({0}) that wasn't expected")]
     UnexpectedMessage(&'static str),
+    #[error("Empty block list requested")]
+    ZeroBlocksInRequest,
+    #[error("Handshake expected")]
+    HandshakeExpected,
+    #[error("More than MAX_ADDRESS_COUNT addresses sent")]
+    AddressListLimitExceeded,
 }
 
 /// Peer state errors (Errors either for an individual peer or for the [`PeerManager`](crate::peer_manager::PeerManager))
@@ -188,14 +195,17 @@ impl BanScore for ProtocolError {
         match self {
             ProtocolError::DifferentNetwork(_, _) => 100,
             ProtocolError::InvalidVersion(_, _) => 100,
-            ProtocolError::InvalidMessage => 100,
             ProtocolError::Unresponsive => 100,
             ProtocolError::LocatorSizeExceeded(_, _) => 20,
             ProtocolError::BlocksRequestLimitExceeded(_, _) => 20,
             ProtocolError::HeadersLimitExceeded(_, _) => 20,
-            ProtocolError::UnknownBlockRequested => 20,
+            ProtocolError::UnknownBlockRequested(_) => 20,
+            ProtocolError::DuplicatedBlockRequest(_) => 20,
             ProtocolError::DisconnectedHeaders => 20,
             ProtocolError::UnexpectedMessage(_) => 20,
+            ProtocolError::ZeroBlocksInRequest => 20,
+            ProtocolError::HandshakeExpected => 100,
+            ProtocolError::AddressListLimitExceeded => 100,
         }
     }
 }
diff --git a/p2p/src/net/default_backend/peer.rs b/p2p/src/net/default_backend/peer.rs
index eef25f8be..ffec1a437 100644
--- a/p2p/src/net/default_backend/peer.rs
+++ b/p2p/src/net/default_backend/peer.rs
@@ -119,7 +119,7 @@ where
                     user_agent,
                 })) = self.socket.recv().await
                 else {
-                    return Err(P2pError::ProtocolError(ProtocolError::InvalidMessage));
+                    return Err(P2pError::ProtocolError(ProtocolError::HandshakeExpected));
                 };
 
                 // Send PeerInfoReceived before sending handshake to remote peer!
@@ -171,7 +171,7 @@ where
                     receiver_address,
                 })) = self.socket.recv().await
                 else {
-                    return Err(P2pError::ProtocolError(ProtocolError::InvalidMessage));
+                    return Err(P2pError::ProtocolError(ProtocolError::HandshakeExpected));
                 };
 
                 self.tx
@@ -514,10 +514,10 @@ mod tests {
             .await
             .unwrap();
 
-        assert_eq!(
+        assert!(matches!(
             handle.await.unwrap(),
-            Err(P2pError::ProtocolError(ProtocolError::InvalidMessage)),
-        );
+            Err(P2pError::ProtocolError(ProtocolError::HandshakeExpected))
+        ),);
     }
 
     #[tokio::test]
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
index e47d4733b..526c686f4 100644
--- a/p2p/src/peer_manager/mod.rs
+++ b/p2p/src/peer_manager/mod.rs
@@ -751,7 +751,7 @@ where
             .ok_or(P2pError::PeerError(PeerError::PeerDoesntExist))?;
         ensure!(
             addresses.len() <= MAX_ADDRESS_COUNT,
-            P2pError::ProtocolError(ProtocolError::InvalidMessage)
+            P2pError::ProtocolError(ProtocolError::AddressListLimitExceeded)
         );
         ensure!(
             peer.expect_addr_list_response,
diff --git a/p2p/src/sync/peer.rs b/p2p/src/sync/peer.rs
index 434ff4762..4205263c0 100644
--- a/p2p/src/sync/peer.rs
+++ b/p2p/src/sync/peer.rs
@@ -198,7 +198,18 @@ where
 
     /// Processes the blocks request.
     async fn handle_block_request(&mut self, block_ids: Vec<Id<Block>>) -> Result<()> {
-        log::debug!("Blocks request from peer {}", self.id());
+        utils::ensure!(
+            !block_ids.is_empty(),
+            P2pError::ProtocolError(ProtocolError::ZeroBlocksInRequest)
+        );
+
+        log::debug!(
+            "Blocks request from peer {}: {}-{} ({})",
+            self.id(),
+            block_ids.first().expect("block_ids is not empty"),
+            block_ids.last().expect("block_ids is not empty"),
+            block_ids.len(),
+        );
 
         if self.is_initial_block_download.load(Ordering::Acquire) {
             log::debug!("Ignoring blocks request because the node is in initial block download");
@@ -225,13 +236,13 @@ where
             .call(move |c| {
                 for id in ids {
                     let index = c.get_block_index(&id)?.ok_or(P2pError::ProtocolError(
-                        ProtocolError::UnknownBlockRequested,
+                        ProtocolError::UnknownBlockRequested(id),
                     ))?;
 
                     if index.block_height() <= best_known_block {
-                        return Err(P2pError::ProtocolError(ProtocolError::UnexpectedMessage(
-                            "Peer requested already known block",
-                        )));
+                        return Err(P2pError::ProtocolError(
+                            ProtocolError::DuplicatedBlockRequest(id),
+                        ));
                     }
                 }
                 Result::<_>::Ok(())
@@ -360,13 +371,27 @@ where
     }
 
     async fn handle_block_announcement(&mut self, header: BlockHeader) -> Result<()> {
-        log::debug!("Block announcement from peer {}: {header:?}", self.id());
+        let block_id = header.block_id();
+        log::debug!(
+            "Block announcement from peer {}: {block_id}: {header:?}",
+            self.id()
+        );
 
         if !self.requested_blocks.is_empty() {
             // We will download this block as part of syncing anyway.
             return Ok(());
         }
 
+        // Do not request the block if it is already known
+        if self
+            .chainstate_handle
+            .call(move |c| c.get_block_index(&block_id))
+            .await??
+            .is_some()
+        {
+            return Ok(());
+        }
+
         let prev_id = *header.prev_block_id();
         if self
             .chainstate_handle
@@ -450,12 +475,22 @@ where
 
         // Remove already requested blocks.
         headers.retain(|h| !self.requested_blocks.contains(&h.get_id()));
+        if headers.is_empty() {
+            return Ok(());
+        }
 
         if headers.len() > *self.p2p_config.max_request_blocks_count {
             self.known_headers = headers.split_off(*self.p2p_config.max_request_blocks_count);
         }
 
         let block_ids: Vec<_> = headers.into_iter().map(|h| h.get_id()).collect();
+        log::debug!(
+            "Request blocks from peer {}: {}-{} ({})",
+            self.id(),
+            block_ids.first().expect("block_ids is not empty"),
+            block_ids.last().expect("block_ids is not empty"),
+            block_ids.len(),
+        );
         self.message_sender.send((
             self.id(),
             SyncMessage::BlockListRequest(BlockListRequest::new(block_ids.clone())),


Test patch:
diff --git a/p2p/src/sync/tests/block_list_request.rs b/p2p/src/sync/tests/block_list_request.rs
index a2bc58f44..3849102b6 100644
--- a/p2p/src/sync/tests/block_list_request.rs
+++ b/p2p/src/sync/tests/block_list_request.rs
@@ -17,7 +17,10 @@ use std::{iter, sync::Arc};
 
 use chainstate::{ban_score::BanScore, BlockSource};
 use chainstate_test_framework::TestFramework;
-use common::{chain::config::create_unit_test_config, primitives::Idable};
+use common::{
+    chain::{config::create_unit_test_config, Block},
+    primitives::{Id, Idable},
+};
 use crypto::random::Rng;
 use p2p_test_utils::{chainstate_subsystem, create_n_blocks};
 use test_utils::random::Seed;
@@ -103,7 +106,8 @@ async fn unknown_blocks(#[case] seed: Seed) {
         .build();
     // Process a block to finish the initial block download.
     tf.make_block_builder().build_and_process().unwrap().unwrap();
-    let unknown_blocks = create_n_blocks(&mut tf, 2).into_iter().map(|b| b.get_id()).collect();
+    let unknown_blocks: Vec<Id<Block>> =
+        create_n_blocks(&mut tf, 2).into_iter().map(|b| b.get_id()).collect();
     let chainstate = chainstate_subsystem(tf.into_chainstate()).await;
 
     let mut handle = SyncManagerHandle::builder()
@@ -115,6 +119,9 @@ async fn unknown_blocks(#[case] seed: Seed) {
     let peer = PeerId::new();
     handle.connect_peer(peer).await;
 
+    let expected_score =
+        P2pError::ProtocolError(ProtocolError::UnknownBlockRequested(unknown_blocks[0]))
+            .ban_score();
     handle.send_message(
         peer,
         SyncMessage::BlockListRequest(BlockListRequest::new(unknown_blocks)),
@@ -122,10 +129,7 @@ async fn unknown_blocks(#[case] seed: Seed) {
 
     let (adjusted_peer, score) = handle.adjust_peer_score_event().await;
     assert_eq!(peer, adjusted_peer);
-    assert_eq!(
-        score,
-        P2pError::ProtocolError(ProtocolError::UnknownBlockRequested).ban_score()
-    );
+    assert_eq!(score, expected_score);
     handle.assert_no_event().await;
 }