Merge pull request #955 from openmina/tweak/pubsub

feat(p2p): More pubsub improvements, docs, cleanup

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

@@ -4,70 +4,119 @@ use mina_p2p_messages::gossip::GossipNetMessageV2;
 use openmina_core::ActionEvent;
 use serde::{Deserialize, Serialize};
 
+/// Actions that can occur within the P2P Network PubSub system.
+///
+/// Managing pubsub streams, handling incoming and outgoing messages,
+/// and maintaining the mesh network topology.
+///
+/// **Common Fields:**
+/// - `peer_id`: The identifier of the peer associated with the action.
+/// - `addr`: The connection address of the peer.
+/// - `stream_id`: The unique identifier of the stream.
+/// - `topic_id`: The identifier of the topic involved in the action.
 #[derive(Serialize, Deserialize, Debug, Clone, ActionEvent)]
 pub enum P2pNetworkPubsubAction {
+    /// Create a new stream, either incoming or outgoing.
+    ///
+    /// **Fields:**
+    /// - `incoming`: Indicates if the stream is incoming (`true`) or outgoing (`false`).
+    /// - `protocol`: The broadcast algorithm used for the stream.
     NewStream {
         incoming: bool,
         peer_id: PeerId,
         addr: ConnectionAddr,
         stream_id: StreamId,
         protocol: BroadcastAlgorithm,
     },
+
+    /// Process incoming raw data from a peer.
+    ///
+    /// **Fields:**
+    /// - `data`: The raw data payload received.
+    /// - `seen_limit`: The limit for tracking seen messages to prevent duplication.
     IncomingData {
         peer_id: PeerId,
         addr: ConnectionAddr,
         stream_id: StreamId,
         data: Data,
         seen_limit: usize,
     },
-    IncomingMessage {
+
+    /// Validate a batch of decoded incoming messages.
+    ValidateIncomingMessages {
         peer_id: PeerId,
-        message: pb::Message,
         seen_limit: usize,
+        addr: ConnectionAddr,
     },
-    IncomingMessageCleanup {
-        peer_id: PeerId,
-    },
-    Graft {
-        peer_id: PeerId,
-        topic_id: String,
-    },
-    Prune {
+
+    /// Handle a fully decoded and validated message received from a peer.
+    ///
+    /// **Fields:**
+    /// - `message`: The decoded protobuf message.
+    /// - `seen_limit`: The limit for tracking seen messages to prevent duplication.
+    IncomingMessage {
         peer_id: PeerId,
-        topic_id: String,
-    },
-    Broadcast {
-        message: Box<GossipNetMessageV2>,
+        message: pb::Message,
+        seen_limit: usize,
     },
+
+    /// Clean up temporary states after processing an incoming message.
+    IncomingMessageCleanup { peer_id: PeerId },
+
+    /// Add a peer to the mesh network for a specific topic.
+    Graft { peer_id: PeerId, topic_id: String },
+
+    /// Remove a peer from the mesh network for a specific topic.
+    Prune { peer_id: PeerId, topic_id: String },
+
+    /// Initiate the broadcasting of a message to all subscribed peers.
+    ///
+    /// **Fields:**
+    /// - `message`: The gossip network message to broadcast.
+    Broadcast { message: Box<GossipNetMessageV2> },
+
+    /// Prepare a message for signing before broadcasting.
+    ///
+    /// **Fields:**
+    /// - `seqno`: The sequence number of the message.
+    /// - `author`: The identifier of the peer authoring the message.
+    /// - `data`: The data payload of the message.
+    /// - `topic`: The topic under which the message is published.
     Sign {
         seqno: u64,
         author: PeerId,
         data: Data,
         topic: String,
     },
+
+    /// An error occured during the signing process.
     #[action_event(level = warn, fields(display(author), display(topic)))]
-    SignError {
-        author: PeerId,
-        topic: String,
-    },
-    BroadcastSigned {
-        signature: Data,
-    },
-    OutgoingMessage {
-        peer_id: PeerId,
-    },
-    OutgoingMessageClear {
-        peer_id: PeerId,
-    },
+    SignError { author: PeerId, topic: String },
+
+    /// Finalize the broadcasting of a signed message by attaching the signature.
+    ///
+    /// **Fields:**
+    /// - `signature`: The cryptographic signature of the message.
+    BroadcastSigned { signature: Data },
+
+    /// Prepare an outgoing message to send to a specific peer.
+    OutgoingMessage { peer_id: PeerId },
+
+    /// Clear the outgoing message state for a specific peer after sending.
+    OutgoingMessageClear { peer_id: PeerId },
+
+    /// An error occured during the sending of an outgoing message.
+    ///
+    /// **Fields:**
+    /// - `msg`: The protobuf message that failed to send.
     #[action_event(level = warn, fields(display(peer_id), debug(msg)))]
-    OutgoingMessageError {
-        msg: pb::Rpc,
-        peer_id: PeerId,
-    },
-    OutgoingData {
-        data: Data,
-        peer_id: PeerId,
-    },
+    OutgoingMessageError { msg: pb::Rpc, peer_id: PeerId },
+
+    /// Send encoded data over an outgoing stream to a specific peer.
+    ///
+    /// **Fields:**
+    /// - `data`: The encoded data to be sent.
+    OutgoingData { data: Data, peer_id: PeerId },
 }
 
 impl From<P2pNetworkPubsubAction> for crate::P2pAction {