fix(p2p): Cleanup pubsub data immediately after it has been processed

Author: tizoc

node/src/action_kind.rs

@@ -404,6 +404,7 @@ pub enum ActionKind {
     P2pNetworkPubsubGraft,
     P2pNetworkPubsubIncomingData,
     P2pNetworkPubsubIncomingMessage,
+    P2pNetworkPubsubIncomingMessageCleanup,
     P2pNetworkPubsubNewStream,
     P2pNetworkPubsubOutgoingData,
     P2pNetworkPubsubOutgoingMessage,
@@ -687,7 +688,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 578;
+    pub const COUNT: u16 = 579;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1892,6 +1893,9 @@ impl ActionKindGet for P2pNetworkPubsubAction {
             Self::NewStream { .. } => ActionKind::P2pNetworkPubsubNewStream,
             Self::IncomingData { .. } => ActionKind::P2pNetworkPubsubIncomingData,
             Self::IncomingMessage { .. } => ActionKind::P2pNetworkPubsubIncomingMessage,
+            Self::IncomingMessageCleanup { .. } => {
+                ActionKind::P2pNetworkPubsubIncomingMessageCleanup
+            }
             Self::Graft { .. } => ActionKind::P2pNetworkPubsubGraft,
             Self::Prune { .. } => ActionKind::P2pNetworkPubsubPrune,
             Self::Broadcast { .. } => ActionKind::P2pNetworkPubsubBroadcast,

p2p/src/network/pubsub/p2p_network_pubsub_actions.rs

@@ -25,6 +25,9 @@ pub enum P2pNetworkPubsubAction {
         message: pb::Message,
         seen_limit: usize,
     },
+    IncomingMessageCleanup {
+        peer_id: PeerId,
+    },
     Graft {
         peer_id: PeerId,
         topic_id: String,

p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs

@@ -163,6 +163,9 @@ impl P2pNetworkPubsubState {
                 pubsub_state.reduce_incoming_message(peer_id, message, seen_limit)?;
 
                 let (dispatcher, global_state) = state_context.into_dispatcher_and_state();
+
+                dispatcher.push(P2pNetworkPubsubAction::IncomingMessageCleanup { peer_id });
+
                 let state: &Self = global_state.substate()?;
                 let config: &P2pConfig = global_state.substate()?;
 
@@ -206,6 +209,26 @@ impl P2pNetworkPubsubState {
                 }
                 Ok(())
             }
+            P2pNetworkPubsubAction::IncomingMessageCleanup { peer_id } => {
+                pubsub_state.incoming_transactions.clear();
+                pubsub_state.incoming_snarks.clear();
+
+                pubsub_state.incoming_transactions.shrink_to(0x20);
+                pubsub_state.incoming_snarks.shrink_to(0x20);
+
+                pubsub_state.incoming_block = None;
+
+                let Some(client_state) = pubsub_state.clients.get_mut(&peer_id) else {
+                    bug_condition!(
+                        "State not found for action P2pNetworkPubsubAction::IncomingMessageCleanup"
+                    );
+                    return Ok(());
+                };
+                client_state.incoming_messages.clear();
+                client_state.incoming_messages.shrink_to(0x20);
+
+                Ok(())
+            }
             // we want to add peer to our mesh
             P2pNetworkPubsubAction::Graft { peer_id, topic_id } => {
                 let Some(state) = pubsub_state
@@ -441,19 +464,6 @@ impl P2pNetworkPubsubState {
         message: Message,
         seen_limit: usize,
     ) -> Result<(), String> {
-        self.incoming_transactions.clear();
-        self.incoming_snarks.clear();
-
-        self.incoming_transactions.shrink_to(0x20);
-        self.incoming_snarks.shrink_to(0x20);
-
-        let Some(state) = self.clients.get_mut(&peer_id) else {
-            bug_condition!("State not found for action P2pNetworkPubsubAction::IncomingMessage");
-            return Ok(());
-        };
-        state.incoming_messages.clear();
-        state.incoming_messages.shrink_to(0x20);
-
         let topic = self.topics.entry(message.topic.clone()).or_default();
 
         if let Some(signature) = &message.signature {