P2p: send a FirstSyncMessageReceived message to the peer manager once instead of sending TransactionSyncMessageTag/BlockSyncMessageTag on each sync message. Add a test to check that the message is actually sent. Some renaming.

ImplOfAnImpl · ImplOfAnImpl · commit 0141c6493e45 · 2026-03-12T20:28:15.000+02:00
diff --git a/dns-server/src/crawler_p2p/crawler_manager/mod.rs b/dns-server/src/crawler_p2p/crawler_manager/mod.rs
@@ -34,7 +34,7 @@ use p2p::{
         PingResponse,
     },
     net::{
-        types::{ConnectivityEvent, PeerManagerMessageOrTag, SyncingEvent},
+        types::{ConnectivityEvent, PeerManagerMessageExt, SyncingEvent},
         ConnectivityService, NetworkingService, SyncingEventReceiver,
     },
     peer_manager::{
@@ -211,12 +211,11 @@ where
     fn handle_conn_message(
         &mut self,
         peer_id: PeerId,
-        message: PeerManagerMessageOrTag,
+        message: PeerManagerMessageExt,
     ) -> p2p::Result<()> {
         let peer_mgr_message = match message {
-            PeerManagerMessageOrTag::PeerManagerMessage(message) => message,
-            PeerManagerMessageOrTag::BlockSyncMessage(_)
-            | PeerManagerMessageOrTag::TransactionSyncMessage(_) => {
+            PeerManagerMessageExt::PeerManagerMessage(message) => message,
+            PeerManagerMessageExt::FirstSyncMessageReceived => {
                 // Ignored
                 return Ok(());
             }
diff --git a/p2p/src/message.rs b/p2p/src/message.rs
@@ -26,7 +26,7 @@ use serialization::{Decode, Encode};
 use crate::types::peer_address::PeerAddress;
 
 #[derive(Debug, Clone, PartialEq, Eq, strum::EnumDiscriminants)]
-#[strum_discriminants(name(BlockSyncMessageTag))]
+#[strum_discriminants(name(BlockSyncMessageTag), derive(strum::EnumIter))]
 pub enum BlockSyncMessage {
     HeaderListRequest(HeaderListRequest),
     BlockListRequest(BlockListRequest),
@@ -42,15 +42,15 @@ pub enum BlockSyncMessage {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, strum::EnumDiscriminants)]
-#[strum_discriminants(name(TransactionSyncMessageTag))]
+#[strum_discriminants(name(TransactionSyncMessageTag), derive(strum::EnumIter))]
 pub enum TransactionSyncMessage {
     NewTransaction(Id<Transaction>),
     TransactionRequest(Id<Transaction>),
     TransactionResponse(TransactionResponse),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, strum::EnumDiscriminants)]
-#[strum_discriminants(name(PeerManagerMessageTag))]
+#[strum_discriminants(name(PeerManagerMessageTag), derive(strum::EnumIter))]
 pub enum PeerManagerMessage {
     AddrListRequest(AddrListRequest),
     AnnounceAddrRequest(AnnounceAddrRequest),
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
@@ -50,7 +50,7 @@ use crate::{
             types::{BackendEvent, Command, PeerEvent},
         },
         types::{
-            services::Services, ConnectivityEvent, PeerInfo, PeerManagerMessageOrTag, SyncingEvent,
+            services::Services, ConnectivityEvent, PeerInfo, PeerManagerMessageExt, SyncingEvent,
         },
     },
     protocol::{ProtocolVersion, SupportedProtocolVersion},
@@ -697,7 +697,7 @@ where
     fn handle_message(
         &mut self,
         peer_id: PeerId,
-        message: PeerManagerMessageOrTag,
+        message: PeerManagerMessageExt,
     ) -> crate::Result<()> {
         // Do not process remaining messages if the peer has been forcibly disconnected (for example, after being banned).
         // Without this check, the backend might send messages to the sync and peer managers after sending the disconnect notification.
diff --git a/p2p/src/net/default_backend/peer.rs b/p2p/src/net/default_backend/peer.rs
@@ -33,7 +33,7 @@ use crate::{
     message::{BlockSyncMessage, TransactionSyncMessage, WillDisconnectMessage},
     net::{
         default_backend::types::{BackendEvent, PeerEvent},
-        types::PeerManagerMessageOrTag,
+        types::PeerManagerMessageExt,
     },
     protocol::{choose_common_protocol_version, ProtocolVersion, SupportedProtocolVersion},
     types::peer_id::PeerId,
@@ -83,6 +83,10 @@ pub struct Peer<T: TransportSocket> {
 
     /// Time getter
     time_getter: TimeGetter,
+
+    /// Will be set to true once at least one BlockSyncMessage or TransactionSyncMessage has been
+    /// received from the peer.
+    sync_message_received: bool,
 }
 
 impl<T> Peer<T>
@@ -115,6 +119,7 @@ where
             node_protocol_version,
             time_getter,
             common_protocol_version: None,
+            sync_message_received: false,
         }
     }
 
@@ -393,17 +398,17 @@ where
     // Note: the channels used by this function to propagate messages to other parts of p2p
     // must be bounded; this is important to prevent DoS attacks.
     async fn handle_socket_msg(
+        &mut self,
         peer_id: PeerId,
         msg: Message,
-        peer_event_sender: &mut mpsc::Sender<PeerEvent>,
-        block_sync_msg_sender: &mut mpsc::Sender<BlockSyncMessage>,
-        transaction_sync_msg_sender: &mut mpsc::Sender<TransactionSyncMessage>,
+        block_sync_msg_sender: &mpsc::Sender<BlockSyncMessage>,
+        transaction_sync_msg_sender: &mpsc::Sender<TransactionSyncMessage>,
     ) -> crate::Result<()> {
         match msg.categorize() {
             CategorizedMessage::Handshake(_) => {
                 log::error!("Peer {peer_id} sent unexpected handshake message");
 
-                peer_event_sender
+                self.peer_event_sender
                     .send(PeerEvent::Misbehaved {
                         error: P2pError::ProtocolError(ProtocolError::UnexpectedMessage(
                             "Unexpected handshake message".to_owned(),
@@ -412,29 +417,35 @@ where
                     .await?;
             }
             CategorizedMessage::PeerManagerMessage(msg) => {
-                peer_event_sender
+                self.peer_event_sender
                     .send(PeerEvent::MessageReceived(
-                        PeerManagerMessageOrTag::PeerManagerMessage(msg),
+                        PeerManagerMessageExt::PeerManagerMessage(msg),
                     ))
                     .await?
             }
             CategorizedMessage::BlockSyncMessage(msg) => {
-                peer_event_sender
-                    .send(PeerEvent::MessageReceived(
-                        PeerManagerMessageOrTag::BlockSyncMessage((&msg).into()),
-                    ))
-                    .await?;
-
-                block_sync_msg_sender.send(msg).await?
+                block_sync_msg_sender.send(msg).await?;
+
+                if !self.sync_message_received {
+                    self.peer_event_sender
+                        .send(PeerEvent::MessageReceived(
+                            PeerManagerMessageExt::FirstSyncMessageReceived,
+                        ))
+                        .await?;
+                    self.sync_message_received = true;
+                }
             }
             CategorizedMessage::TransactionSyncMessage(msg) => {
-                peer_event_sender
-                    .send(PeerEvent::MessageReceived(
-                        PeerManagerMessageOrTag::TransactionSyncMessage((&msg).into()),
-                    ))
-                    .await?;
-
-                transaction_sync_msg_sender.send(msg).await?
+                transaction_sync_msg_sender.send(msg).await?;
+
+                if !self.sync_message_received {
+                    self.peer_event_sender
+                        .send(PeerEvent::MessageReceived(
+                            PeerManagerMessageExt::FirstSyncMessageReceived,
+                        ))
+                        .await?;
+                    self.sync_message_received = true;
+                }
             }
         }
 
@@ -507,12 +518,11 @@ where
                 event = self.socket.recv(), if sync_msg_senders_opt.is_some() => match event {
                     Ok(message) => {
                         let sync_msg_senders = sync_msg_senders_opt.as_mut().expect("sync_msg_senders_opt is some");
-                        Self::handle_socket_msg(
+                        self.handle_socket_msg(
                             self.peer_id,
                             message,
-                            &mut self.peer_event_sender,
-                            &mut sync_msg_senders.0,
-                            &mut sync_msg_senders.1,
+                            &sync_msg_senders.0,
+                            &sync_msg_senders.1,
                         ).await?;
                     }
                     Err(err) => {
diff --git a/p2p/src/net/default_backend/types.rs b/p2p/src/net/default_backend/types.rs
@@ -32,7 +32,7 @@ use crate::{
         BlockSyncMessage, HeaderList, HeaderListRequest, PeerManagerMessage, PingRequest,
         PingResponse, TransactionResponse, TransactionSyncMessage, WillDisconnectMessage,
     },
-    net::types::{services::Services, PeerManagerMessageOrTag},
+    net::types::{services::Services, PeerManagerMessageExt},
     protocol::{ProtocolVersion, SupportedProtocolVersion},
     types::{peer_address::PeerAddress, peer_id::PeerId},
 };
@@ -119,7 +119,7 @@ pub enum PeerEvent {
     ConnectionClosed,
 
     /// Message received from remote
-    MessageReceived(PeerManagerMessageOrTag),
+    MessageReceived(PeerManagerMessageExt),
 
     /// Protocol violation
     Misbehaved { error: P2pError },
diff --git a/p2p/src/net/types.rs b/p2p/src/net/types.rs
@@ -27,8 +27,7 @@ use tokio::sync::mpsc::Receiver;
 use crate::{
     error::ConnectionValidationError,
     message::{
-        BlockSyncMessage, BlockSyncMessageTag, PeerManagerMessage, PeerManagerMessageTag,
-        TransactionSyncMessage, TransactionSyncMessageTag,
+        BlockSyncMessage, PeerManagerMessage, PeerManagerMessageTag, TransactionSyncMessage,
     },
     protocol::SupportedProtocolVersion,
     types::{peer_address::PeerAddress, peer_id::PeerId},
@@ -166,11 +165,9 @@ impl Display for PeerInfo {
 #[derive(Debug)]
 pub enum ConnectivityEvent {
     /// A message received from a peer.
-    ///
-    /// Note that only a message tag is present here for block and transaction sync messages.
     Message {
         peer_id: PeerId,
-        message: PeerManagerMessageOrTag,
+        message: PeerManagerMessageExt,
     },
 
     /// Outbound connection accepted
@@ -248,37 +245,34 @@ pub enum ConnectivityEvent {
     },
 }
 
-/// Either a full `PeerManagerMessage` or, if it's a sync message, the corresponding tag.
-#[derive(Debug, Clone)]
-pub enum PeerManagerMessageOrTag {
+#[derive(Debug)]
+pub enum PeerManagerMessageExt {
+    // The complete PeerManagerMessage
     PeerManagerMessage(PeerManagerMessage),
-    BlockSyncMessage(BlockSyncMessageTag),
-    TransactionSyncMessage(TransactionSyncMessageTag),
+
+    // An indicator that the first sync message (i.e. BlockSyncMessage or TransactionSyncMessage)
+    // has been received from the peer.
+    FirstSyncMessageReceived,
 }
 
-impl From<PeerManagerMessage> for PeerManagerMessageOrTag {
+impl From<PeerManagerMessage> for PeerManagerMessageExt {
     fn from(value: PeerManagerMessage) -> Self {
         Self::PeerManagerMessage(value)
     }
 }
 
+/// Tag type for `PeerManagerMessageExt`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum ConnectivityEventMessageTag {
+pub enum PeerManagerMessageExtTag {
     PeerManagerMessage(PeerManagerMessageTag),
-    BlockSyncMessage(BlockSyncMessageTag),
-    TransactionSyncMessage(TransactionSyncMessageTag),
+    FirstSyncMessageReceived,
 }
 
-impl From<&'_ PeerManagerMessageOrTag> for ConnectivityEventMessageTag {
-    fn from(value: &'_ PeerManagerMessageOrTag) -> Self {
+impl From<&'_ PeerManagerMessageExt> for PeerManagerMessageExtTag {
+    fn from(value: &'_ PeerManagerMessageExt) -> Self {
         match value {
-            PeerManagerMessageOrTag::PeerManagerMessage(msg) => {
-                Self::PeerManagerMessage(msg.into())
-            }
-            PeerManagerMessageOrTag::BlockSyncMessage(tag) => Self::BlockSyncMessage(*tag),
-            PeerManagerMessageOrTag::TransactionSyncMessage(tag) => {
-                Self::TransactionSyncMessage(*tag)
-            }
+            PeerManagerMessageExt::PeerManagerMessage(msg) => Self::PeerManagerMessage(msg.into()),
+            PeerManagerMessageExt::FirstSyncMessageReceived => Self::FirstSyncMessageReceived,
         }
     }
 }
diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs
@@ -63,8 +63,7 @@ use crate::{
     net::{
         types::{
             services::{Service, Services},
-            ConnectivityEvent, ConnectivityEventMessageTag, PeerInfo, PeerManagerMessageOrTag,
-            PeerRole,
+            ConnectivityEvent, PeerInfo, PeerManagerMessageExt, PeerManagerMessageExtTag, PeerRole,
         },
         ConnectivityService, NetworkingService,
     },
@@ -1563,9 +1562,9 @@ where
         }
     }
 
-    fn handle_incoming_message(&mut self, peer_id: PeerId, message: PeerManagerMessageOrTag) {
+    fn handle_incoming_message(&mut self, peer_id: PeerId, message: PeerManagerMessageExt) {
         let is_disconnection_message = match &message {
-            PeerManagerMessageOrTag::PeerManagerMessage(msg) => match msg {
+            PeerManagerMessageExt::PeerManagerMessage(msg) => match msg {
                 PeerManagerMessage::WillDisconnect(_) => true,
 
                 PeerManagerMessage::AddrListRequest(_)
@@ -1574,8 +1573,7 @@ where
                 | PeerManagerMessage::AddrListResponse(_)
                 | PeerManagerMessage::PingResponse(_) => false,
             },
-            PeerManagerMessageOrTag::BlockSyncMessage(_)
-            | PeerManagerMessageOrTag::TransactionSyncMessage(_) => false,
+            PeerManagerMessageExt::FirstSyncMessageReceived => false,
         };
 
         // Note: `PeerContext` must always exist when an incoming message arrives, and the individual
@@ -1595,10 +1593,10 @@ where
             return;
         }
 
-        let message_tag: ConnectivityEventMessageTag = (&message).into();
+        let message_tag: PeerManagerMessageExtTag = (&message).into();
 
         match message {
-            PeerManagerMessageOrTag::PeerManagerMessage(msg) => match msg {
+            PeerManagerMessageExt::PeerManagerMessage(msg) => match msg {
                 PeerManagerMessage::AddrListRequest(_) => {
                     self.handle_addr_list_request(peer_id);
                 }
@@ -1618,8 +1616,7 @@ where
                     self.handle_will_disconnect_messgae(peer_id, msg);
                 }
             },
-            PeerManagerMessageOrTag::BlockSyncMessage(_)
-            | PeerManagerMessageOrTag::TransactionSyncMessage(_) => {}
+            PeerManagerMessageExt::FirstSyncMessageReceived => {}
         };
 
         if let Some(o) = self.observer.as_mut() {
@@ -2481,7 +2478,7 @@ pub trait Observer {
 
     // This will be called after `ConnectivityEvent::Message` has been handled by
     // the peer manager.
-    fn message_received(&mut self, peer_id: PeerId, message_tag: ConnectivityEventMessageTag);
+    fn message_received(&mut self, peer_id: PeerId, message_tag: PeerManagerMessageExtTag);
 }
 
 pub trait PeerManagerInterface {
diff --git a/p2p/src/peer_manager/tests/unsuccessful_connection_counter_update.rs b/p2p/src/peer_manager/tests/unsuccessful_connection_counter_update.rs
@@ -44,12 +44,9 @@ use crate::{
     config::P2pConfig,
     disconnection_reason::DisconnectionReason,
     error::{DialError, P2pError},
-    message::BlockSyncMessageTag,
     net::{
         default_backend::types::{Command, Message},
-        types::{
-            ConnectivityEvent, ConnectivityEventMessageTag, PeerManagerMessageOrTag, PeerRole,
-        },
+        types::{ConnectivityEvent, PeerManagerMessageExt, PeerManagerMessageExtTag, PeerRole},
     },
     peer_manager::{
         self,
@@ -745,9 +742,7 @@ async fn auto_connection_without_peer_activity(
             conn_event_sender
                 .send(ConnectivityEvent::Message {
                     peer_id,
-                    message: PeerManagerMessageOrTag::BlockSyncMessage(
-                        BlockSyncMessageTag::HeaderListRequest,
-                    ),
+                    message: PeerManagerMessageExt::FirstSyncMessageReceived,
                 })
                 .unwrap();
 
@@ -756,9 +751,7 @@ async fn auto_connection_without_peer_activity(
                 peer_mgr_notification,
                 PeerManagerNotification::MessageReceived {
                     peer_id,
-                    message_tag: ConnectivityEventMessageTag::BlockSyncMessage(
-                        BlockSyncMessageTag::HeaderListRequest
-                    )
+                    message_tag: PeerManagerMessageExtTag::FirstSyncMessageReceived
                 }
             );
         }
diff --git a/p2p/src/tests/helpers/mod.rs b/p2p/src/tests/helpers/mod.rs
diff --git a/p2p/src/tests/helpers/test_node.rs b/p2p/src/tests/helpers/test_node.rs
diff --git a/p2p/src/tests/mod.rs b/p2p/src/tests/mod.rs
diff --git a/p2p/src/tests/peer_mgr_events.rs b/p2p/src/tests/peer_mgr_events.rs
