fix(p2p): Avoid duplicating incoming messages during validation

Author: tizoc

node/src/action_kind.rs

@@ -413,8 +413,9 @@ pub enum ActionKind {
     P2pNetworkPubsubPrune,
     P2pNetworkPubsubSign,
     P2pNetworkPubsubSignError,
-    P2pNetworkPubsubEffectfulIncomingData,
+    P2pNetworkPubsubValidateIncomingMessages,
     P2pNetworkPubsubEffectfulSign,
+    P2pNetworkPubsubEffectfulValidateIncomingMessages,
     P2pNetworkRpcHeartbeatSend,
     P2pNetworkRpcIncomingData,
     P2pNetworkRpcIncomingMessage,
@@ -688,7 +689,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 579;
+    pub const COUNT: u16 = 580;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1892,6 +1893,9 @@ impl ActionKindGet for P2pNetworkPubsubAction {
         match self {
             Self::NewStream { .. } => ActionKind::P2pNetworkPubsubNewStream,
             Self::IncomingData { .. } => ActionKind::P2pNetworkPubsubIncomingData,
+            Self::ValidateIncomingMessages { .. } => {
+                ActionKind::P2pNetworkPubsubValidateIncomingMessages
+            }
             Self::IncomingMessage { .. } => ActionKind::P2pNetworkPubsubIncomingMessage,
             Self::IncomingMessageCleanup { .. } => {
                 ActionKind::P2pNetworkPubsubIncomingMessageCleanup
@@ -1994,7 +1998,9 @@ impl ActionKindGet for P2pNetworkPubsubEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {
             Self::Sign { .. } => ActionKind::P2pNetworkPubsubEffectfulSign,
-            Self::IncomingData { .. } => ActionKind::P2pNetworkPubsubEffectfulIncomingData,
+            Self::ValidateIncomingMessages { .. } => {
+                ActionKind::P2pNetworkPubsubEffectfulValidateIncomingMessages
+            }
         }
     }
 }

p2p/src/network/pubsub/p2p_network_pubsub_actions.rs

@@ -42,7 +42,14 @@ pub enum P2pNetworkPubsubAction {
         seen_limit: usize,
     },
 
-    /// Handle a fully decoded message received from a peer.
+    /// Validate a batch of decoded incoming messages.
+    ValidateIncomingMessages {
+        peer_id: PeerId,
+        seen_limit: usize,
+        addr: ConnectionAddr,
+    },
+
+    /// Handle a fully decoded and validated message received from a peer.
     ///
     /// **Fields:**
     /// - `message`: The decoded protobuf message.

p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs

@@ -8,7 +8,7 @@ use redux::{Dispatcher, Timestamp};
 use crate::{
     channels::{snark::P2pChannelsSnarkAction, transaction::P2pChannelsTransactionAction},
     peer::P2pPeerAction,
-    Data, P2pConfig, P2pNetworkYamuxAction, P2pState, PeerId,
+    Data, P2pConfig, P2pNetworkYamuxAction, PeerId,
 };
 
 use super::{
@@ -135,18 +135,31 @@ impl P2pNetworkPubsubState {
             } => {
                 pubsub_state.reduce_incoming_data(&peer_id, data, meta.time())?;
 
-                let (dispatcher, state) = state_context.into_dispatcher_and_state();
-                let p2p_state: &P2pState = state.substate()?;
-                let state = &p2p_state.network.scheduler.broadcast_state;
-                let Some(state) = state.clients.get(&peer_id) else {
+                let dispatcher = state_context.into_dispatcher();
+
+                dispatcher.push(P2pNetworkPubsubAction::ValidateIncomingMessages {
+                    peer_id,
+                    seen_limit,
+                    addr,
+                });
+
+                Ok(())
+            }
+            P2pNetworkPubsubAction::ValidateIncomingMessages {
+                peer_id,
+                seen_limit,
+                addr,
+            } => {
+                let Some(state) = pubsub_state.clients.get_mut(&peer_id) else {
                     // TODO: investigate, cannot reproduce this
                     // bug_condition!("{:?} not found in state.clients", peer_id);
                     return Ok(());
                 };
+                let messages = std::mem::take(&mut state.incoming_messages);
+
+                let dispatcher = state_context.into_dispatcher();
 
-                // TODO: try to reuse data instead of cloning all message
-                let messages = state.incoming_messages.clone();
-                dispatcher.push(P2pNetworkPubsubEffectfulAction::IncomingData {
+                dispatcher.push(P2pNetworkPubsubEffectfulAction::ValidateIncomingMessages {
                     peer_id,
                     seen_limit,
                     addr,

p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_actions.rs

@@ -2,14 +2,29 @@ use crate::{pubsub::pb::Message, ConnectionAddr, P2pState, PeerId};
 use openmina_core::ActionEvent;
 use serde::{Deserialize, Serialize};
 
+/// Eeffectful actions within the P2P Network PubSub system.
 #[derive(Serialize, Deserialize, Debug, Clone, ActionEvent)]
 pub enum P2pNetworkPubsubEffectfulAction {
+    /// Initiate the signing of a message before broadcasting.
+    ///
+    /// **Fields:**
+    /// - `author`: The identifier of the peer authoring the message.
+    /// - `topic`: The topic under which the message is published.
+    /// - `message`: The protobuf message to be signed.
     Sign {
         author: PeerId,
         topic: String,
         message: Message,
     },
-    IncomingData {
+
+    /// Validate a batch of incoming messages from a peer.
+    ///
+    /// **Fields:**
+    /// - `peer_id`: The identifier of the peer sending the messages.
+    /// - `seen_limit`: The limit for tracking seen messages to prevent duplication.
+    /// - `addr`: The connection address of the peer.
+    /// - `messages`: Decoded protobuf messages.
+    ValidateIncomingMessages {
         peer_id: PeerId,
         seen_limit: usize,
         addr: ConnectionAddr,

p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_effects.rs

@@ -30,7 +30,7 @@ impl P2pNetworkPubsubEffectfulAction {
                     store.dispatch(P2pNetworkPubsubAction::BroadcastSigned { signature });
                 }
             }
-            P2pNetworkPubsubEffectfulAction::IncomingData {
+            P2pNetworkPubsubEffectfulAction::ValidateIncomingMessages {
                 peer_id,
                 seen_limit,
                 addr,
@@ -73,6 +73,9 @@ impl P2pNetworkPubsubEffectfulAction {
                             error: P2pNetworkConnectionError::PubSubError(error.to_string()),
                         });
 
+                        // TODO: should this error short-circuit the whole batch?
+                        // if yes, shouldn't every message be verified first before
+                        // dispatching `P2pNetworkPubsubAction::IncomingMessage`?
                         return;
                     }
                 }