Merge pull request #1309 from mintlayer/attempt_to_fix_spuriously_failing_test

Attempt to fix spuriously failing test "dont_make_announcements_while_blocks_are_being_sent".
Also, PR #1316 was merged into this branch.

Author: ImplOfAnImpl

p2p/backend-test-suite/src/ban.rs

@@ -133,12 +133,14 @@ where
         };
         messaging_handle_2
             .send_message(peer, SyncMessage::HeaderList(HeaderList::new(Vec::new())))
+            .await
             .unwrap();
         messaging_handle_2
             .send_message(
                 peer,
                 SyncMessage::HeaderList(HeaderList::new(vec![block.header().clone()])),
             )
+            .await
             .unwrap();
     });
 

p2p/backend-test-suite/src/block_announcement.rs

@@ -94,6 +94,7 @@ where
             peer2.peer_id,
             SyncMessage::HeaderList(HeaderList::new(vec![block.header().clone()])),
         )
+        .await
         .unwrap();
 
     let mut sync_msg_rx_2 = match sync2.poll_next().await.unwrap() {
@@ -130,6 +131,7 @@ where
             peer1.peer_id,
             SyncMessage::HeaderList(HeaderList::new(vec![block.header().clone()])),
         )
+        .await
         .unwrap();
 
     let mut sync_msg_rx_1 = match sync1.poll_next().await.unwrap() {
@@ -236,6 +238,7 @@ where
             peer2.peer_id,
             SyncMessage::HeaderList(HeaderList::new(vec![block.header().clone()])),
         )
+        .await
         .unwrap();
 
     shutdown.store(true);

p2p/src/error.rs

@@ -49,8 +49,15 @@ pub enum ProtocolError {
     DuplicatedBlockRequest(Id<Block>),
     #[error("Headers aren't connected")]
     DisconnectedHeaders,
-    #[error("Received a message ({0}) that wasn't expected")]
+    #[error("Peer sent a message ({0}) that wasn't expected")]
     UnexpectedMessage(String),
+    #[error("Peer sent a block ({0}) that wasn't requested")]
+    UnsolicitedBlockReceived(Id<Block>),
+    #[error("Peer sent block {expected_block_id} while it was expected to send {actual_block_id}")]
+    BlocksReceivedInWrongOrder {
+        expected_block_id: Id<Block>,
+        actual_block_id: Id<Block>,
+    },
     #[error("Empty block list requested")]
     ZeroBlocksInRequest,
     #[error("Handshake expected")]
@@ -227,6 +234,11 @@ impl BanScore for ProtocolError {
             ProtocolError::DuplicatedBlockRequest(_) => 20,
             ProtocolError::DisconnectedHeaders => 20,
             ProtocolError::UnexpectedMessage(_) => 20,
+            ProtocolError::UnsolicitedBlockReceived(_) => 20,
+            ProtocolError::BlocksReceivedInWrongOrder {
+                expected_block_id: _,
+                actual_block_id: _,
+            } => 20,
             ProtocolError::ZeroBlocksInRequest => 20,
             ProtocolError::HandshakeExpected => 100,
             ProtocolError::AddressListLimitExceeded => 100,

p2p/src/net/default_backend/mod.rs

@@ -142,8 +142,10 @@ where
     }
 }
 
+#[async_trait]
 impl MessagingService for MessagingHandle {
-    fn send_message(&mut self, peer_id: PeerId, message: SyncMessage) -> crate::Result<()> {
+    async fn send_message(&mut self, peer_id: PeerId, message: SyncMessage) -> crate::Result<()> {
+        // Note: send_message was made async to be able to use a bounded channel in tests.
         Ok(self.command_sender.send(types::Command::SendMessage {
             peer_id,
             message: message.into(),

p2p/src/net/mod.rs

@@ -122,9 +122,10 @@ where
 }
 
 /// An interface for sending messages and announcements to peers.
+#[async_trait]
 pub trait MessagingService: Clone {
     /// Sends a message to the peer.
-    fn send_message(&mut self, peer: PeerId, message: SyncMessage) -> crate::Result<()>;
+    async fn send_message(&mut self, peer: PeerId, message: SyncMessage) -> crate::Result<()>;
 }
 
 #[async_trait]

p2p/src/sync/peer_v1.rs

@@ -103,7 +103,7 @@ struct IncomingDataState {
     /// requested the blocks for.
     pending_headers: Vec<SignedBlockHeader>,
     /// A list of blocks that we requested from this peer.
-    requested_blocks: BTreeSet<Id<Block>>,
+    requested_blocks: VecDeque<Id<Block>>,
     /// The id of the best block header that we've received from the peer and that we also have.
     /// This includes headers received by any means, e.g. via HeaderList messages, as part
     /// of a locator during peer's header requests, via block responses.
@@ -158,7 +158,7 @@ where
             time_getter,
             incoming: IncomingDataState {
                 pending_headers: Vec::new(),
-                requested_blocks: BTreeSet::new(),
+                requested_blocks: VecDeque::new(),
                 peers_best_block_that_we_have: None,
                 singular_unconnected_headers_count: 0,
             },
@@ -225,15 +225,15 @@ where
         }
     }
 
-    fn send_message(&mut self, message: SyncMessage) -> Result<()> {
-        self.messaging_handle.send_message(self.id(), message)
+    async fn send_message(&mut self, message: SyncMessage) -> Result<()> {
+        self.messaging_handle.send_message(self.id(), message).await
     }
 
-    fn send_headers(&mut self, headers: HeaderList) -> Result<()> {
+    async fn send_headers(&mut self, headers: HeaderList) -> Result<()> {
         if let Some(last_header) = headers.headers().last() {
             self.outgoing.best_sent_block_header = Some(last_header.block_id().into());
         }
-        self.send_message(SyncMessage::HeaderList(headers))
+        self.send_message(SyncMessage::HeaderList(headers)).await
     }
 
     async fn handle_new_tip(&mut self, new_tip_id: &Id<Block>) -> Result<()> {
@@ -290,23 +290,17 @@ where
                     .await?;
 
                 if headers.is_empty() {
-                    log::warn!(
-                        concat!(
-                            "[peer id = {}] Got new tip event with block id {}, ",
-                            "but there is nothing to send"
-                        ),
+                    log::debug!(
+                        "[peer id = {}] Got new tip event with block id {}, but there is nothing to send",
                         self.id(),
                         new_tip_id,
                     );
                 } else if best_block_id != new_tip_id {
                     // If we got here, another "new tip" event should be generated soon,
                     // so we may ignore this one (and it makes sense to ignore it to avoid sending
                     // the same header list multiple times).
-                    log::warn!(
-                        concat!(
-                            "[peer id = {}] Got new tip event with block id {}, ",
-                            "but the tip has changed since then to {}"
-                        ),
+                    log::info!(
+                        "[peer id = {}] Got new tip event with block id {}, but the tip has changed since then to {}",
                         self.id(),
                         new_tip_id,
                         best_block_id
@@ -317,7 +311,7 @@ where
                         self.id(),
                         headers.len()
                     );
-                    return self.send_headers(HeaderList::new(headers));
+                    return self.send_headers(HeaderList::new(headers)).await;
                 }
             } else {
                 // Note: if we got here, then we haven't received a single header request or
@@ -346,7 +340,7 @@ where
                     && self.common_services.has_service(Service::Transactions)
                 {
                     self.add_known_transaction(txid);
-                    self.send_message(SyncMessage::NewTransaction(txid))
+                    self.send_message(SyncMessage::NewTransaction(txid)).await
                 } else {
                     Ok(())
                 }
@@ -357,9 +351,6 @@ where
     async fn request_headers(&mut self) -> Result<()> {
         let locator = self.chainstate_handle.call(|this| Ok(this.get_locator()?)).await?;
         if locator.len() > *self.p2p_config.protocol_config.msg_max_locator_count {
-            // Note: msg_max_locator_count is not supposed to be configurable outside of tests,
-            // so we should never get here in production code. Moreover, currently it's not
-            // modified even in tests. TODO: make it a constant.
             log::warn!(
                 "[peer id = {}] Sending locator of the length {}, which exceeds the maximum length {:?}",
                 self.id(),
@@ -371,7 +362,8 @@ where
         log::debug!("[peer id = {}] Sending header list request", self.id());
         self.send_message(SyncMessage::HeaderListRequest(HeaderListRequest::new(
             locator,
-        )))?;
+        )))
+        .await?;
 
         self.peer_activity
             .set_expecting_headers_since(Some(self.time_getter.get_time()));
@@ -445,7 +437,7 @@ where
         // all headers that were available at the moment.
         self.have_sent_all_headers = headers.len() < header_count_limit;
 
-        self.send_headers(HeaderList::new(headers))
+        self.send_headers(HeaderList::new(headers)).await
     }
 
     /// Processes the blocks request.
@@ -719,7 +711,7 @@ where
                 .await?;
         }
 
-        self.request_blocks(new_block_headers)
+        self.request_blocks(new_block_headers).await
     }
 
     async fn handle_block_response(&mut self, block: Block) -> Result<()> {
@@ -734,10 +726,28 @@ where
         // The code below will set it again if needed.
         self.peer_activity.set_expecting_blocks_since(None);
 
-        if self.incoming.requested_blocks.take(&block.get_id()).is_none() {
-            return Err(P2pError::ProtocolError(ProtocolError::UnexpectedMessage(
-                "block response".to_owned(),
-            )));
+        if self.incoming.requested_blocks.front().is_some_and(|id| id == &block.get_id()) {
+            self.incoming.requested_blocks.pop_front();
+        } else {
+            let idx = self.incoming.requested_blocks.iter().position(|id| id == &block.get_id());
+            // Note: we treat wrongly ordered blocks in the same way as unsolicited ones, i.e.
+            // we don't remove their ids from the list.
+            if idx.is_some() {
+                return Err(P2pError::ProtocolError(
+                    ProtocolError::BlocksReceivedInWrongOrder {
+                        expected_block_id: *self
+                            .incoming
+                            .requested_blocks
+                            .front()
+                            .expect("The deque is known to be non-empty"),
+                        actual_block_id: block.get_id(),
+                    },
+                ));
+            } else {
+                return Err(P2pError::ProtocolError(
+                    ProtocolError::UnsolicitedBlockReceived(block.get_id()),
+                ));
+            }
         }
 
         let block = self.chainstate_handle.call(|c| Ok(c.preliminary_block_check(block)?)).await?;
@@ -798,7 +808,7 @@ where
                 self.request_headers().await?;
             } else {
                 // Download remaining blocks.
-                self.request_blocks(headers)?;
+                self.request_blocks(headers).await?;
             }
         } else {
             // We expect additional blocks from the peer, update the timestamp.
@@ -821,7 +831,7 @@ where
             None => TransactionResponse::NotFound(id),
         };
 
-        self.send_message(SyncMessage::TransactionResponse(res))?;
+        self.send_message(SyncMessage::TransactionResponse(res)).await?;
 
         Ok(())
     }
@@ -905,7 +915,7 @@ where
         }
 
         if !(self.mempool_handle.call(move |m| m.contains_transaction(&tx)).await?) {
-            self.send_message(SyncMessage::TransactionRequest(tx))?;
+            self.send_message(SyncMessage::TransactionRequest(tx)).await?;
             assert!(self.announced_transactions.insert(tx));
         }
 
@@ -916,7 +926,7 @@ where
     ///
     /// The number of blocks requested equals `P2pConfig::requested_blocks_limit`, the remaining
     /// headers are stored in the peer context.
-    fn request_blocks(&mut self, mut headers: Vec<SignedBlockHeader>) -> Result<()> {
+    async fn request_blocks(&mut self, mut headers: Vec<SignedBlockHeader>) -> Result<()> {
         debug_assert!(self.incoming.pending_headers.is_empty());
         debug_assert!(self.incoming.requested_blocks.is_empty());
         debug_assert!(!headers.is_empty());
@@ -936,7 +946,8 @@ where
         );
         self.send_message(SyncMessage::BlockListRequest(BlockListRequest::new(
             block_ids.clone(),
-        )))?;
+        )))
+        .await?;
         self.incoming.requested_blocks.extend(block_ids);
 
         self.peer_activity.set_expecting_blocks_since(Some(self.time_getter.get_time()));
@@ -945,7 +956,7 @@ where
     }
 
     async fn send_block(&mut self, id: Id<Block>) -> Result<()> {
-        let (block, index) = self
+        let (block, block_index) = self
             .chainstate_handle
             .call(move |c| {
                 let index = c.get_block_index(&id);
@@ -960,14 +971,27 @@ where
         // to delete block indices of missing blocks when resetting their failure flags).
         // P2p should handle such situations correctly (see issue #1033 for more details).
         let block = block?.unwrap_or_else(|| panic!("Unknown block requested: {id}"));
-        self.outgoing.best_sent_block = index?;
+        let block_index = block_index?.expect("Block index must exist");
+
+        let old_best_sent_block_id = self.outgoing.best_sent_block.as_ref().map(|idx| {
+            let id: Id<GenBlock> = (*idx.block_id()).into();
+            id
+        });
+        let new_best_sent_block_id = self
+            .chainstate_handle
+            .call(move |c| choose_peers_best_block(c, old_best_sent_block_id, Some(id.into())))
+            .await?;
+
+        if new_best_sent_block_id == Some(id.into()) {
+            self.outgoing.best_sent_block = Some(block_index);
+        }
 
         log::debug!(
             "[peer id = {}] Sending block with id = {} to the peer",
             self.id(),
             block.get_id()
         );
-        self.send_message(SyncMessage::BlockResponse(BlockResponse::new(block)))
+        self.send_message(SyncMessage::BlockResponse(BlockResponse::new(block))).await
     }
 
     async fn disconnect_if_stalling(&mut self) -> Result<()> {