doc(p2p): Document pubsub actions and state

Author: tizoc

p2p/src/network/pubsub/p2p_network_pubsub_actions.rs

@@ -4,70 +4,112 @@ 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,
     },
+
+    /// Handle a fully decoded 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,
         message: pb::Message,
         seen_limit: usize,
     },
-    IncomingMessageCleanup {
-        peer_id: PeerId,
-    },
-    Graft {
-        peer_id: PeerId,
-        topic_id: String,
-    },
-    Prune {
-        peer_id: PeerId,
-        topic_id: String,
-    },
-    Broadcast {
-        message: Box<GossipNetMessageV2>,
-    },
+
+    /// 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 {

p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs

@@ -548,7 +548,7 @@ impl P2pNetworkPubsubState {
     }
 
     /// Processes incoming data from a peer, handling subscriptions, control messages,
-    /// and message dissemination within the P2P pubsub system.
+    /// and message broadcasting within the P2P pubsub system.
     fn reduce_incoming_data(
         &mut self,
         peer_id: &PeerId,

p2p/src/network/pubsub/p2p_network_pubsub_state.rs

@@ -14,17 +14,49 @@ use serde::{Deserialize, Serialize};
 
 pub const IWANT_TIMEOUT_DURATION: Duration = Duration::from_secs(5);
 
+/// State of the P2P Network PubSub system.
+///
+/// This struct maintains information about connected peers, message sequencing,
+/// message caching, and topic subscriptions. It handles incoming and outgoing
+/// messages, manages the mesh network topology, and ensures efficient message
+/// broadcasting across the network.
 #[derive(Default, Serialize, Deserialize, Debug, Clone)]
 pub struct P2pNetworkPubsubState {
+    /// State of each connected peer.
     pub clients: BTreeMap<PeerId, P2pNetworkPubsubClientState>,
+
+    /// Current message sequence number.
+    ///
+    /// Increments with each new message to ensure proper ordering and uniqueness.
     pub seq: u64,
+
+    /// Messages awaiting cryptographic signing.
     pub to_sign: VecDeque<pb::Message>,
+
+    /// Recently seen message identifiers to prevent duplication.
+    ///
+    /// Keeps a limited history of message signatures to avoid processing
+    /// the same message multiple times.
     pub seen: VecDeque<Vec<u8>>,
+
+    /// Cache of published messages for efficient retrieval and broadcasting.
+    ///
+    /// For quick access and reducing redundant data transmission across peers.
     pub mcache: P2pNetworkPubsubMessageCache,
+
+    /// Incoming block from a peer, if any.
     pub incoming_block: Option<(PeerId, Arc<v2::MinaBlockBlockStableV2>)>,
+
+    /// Incoming transactions from peers along with their nonces.
     pub incoming_transactions: Vec<(Transaction, u32)>,
+
+    /// Incoming snarks from peers along with their nonces.
     pub incoming_snarks: Vec<(Snark, u32)>,
+
+    /// Topics and their subscribed peers.
     pub topics: BTreeMap<String, BTreeMap<PeerId, P2pNetworkPubsubClientTopicState>>,
+
+    /// `iwant` requests, tracking the number of times peers have expressed interest in specific messages.
     pub iwant: VecDeque<P2pNetworkPubsubIwantRequestCount>,
 }
 
@@ -86,14 +118,45 @@ impl P2pNetworkPubsubState {
     }
 }
 
+/// State of a pubsub client connected to a peer.
+///
+/// This struct maintains essential information about the client's protocol,
+/// connection details, message buffers, and caching mechanisms. It facilitates
+/// efficient message handling and broadcasting within the pubsub system.
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct P2pNetworkPubsubClientState {
+    /// Broadcast algorithm used for this client.
     pub protocol: BroadcastAlgorithm,
+
+    /// Connection address of the peer.
     pub addr: ConnectionAddr,
+
+    /// Outgoing stream identifier, if any.
+    ///
+    /// - `Some(StreamId)`: Indicates an active outgoing stream.
+    /// - `None`: No outgoing stream is currently established.
     pub outgoing_stream_id: Option<StreamId>,
+
+    /// Current RPC message being constructed or processed.
+    ///
+    /// - `subscriptions`: List of subscription options for various topics.
+    /// - `publish`: Messages queued for publishing.
+    /// - `control`: Control commands for managing the mesh network.
     pub message: pb::Rpc,
+
+    /// Cache of recently published messages.
     pub cache: P2pNetworkPubsubRecentlyPublishCache,
+
+    /// Buffer for incoming data fragments.
+    ///
+    /// Stores partial data received from peers, facilitating the assembly of complete
+    /// messages when all fragments are received.
     pub buffer: Vec<u8>,
+
+    /// Collection of incoming messages from the peer.
+    ///
+    /// Holds fully decoded `pb::Message` instances received from the peer,
+    /// ready for further handling such as validation, caching, and broadcasting.
     pub incoming_messages: Vec<pb::Message>,
 }