feat(p2p/libp2p): rebroadcast gossip messages received on webrtc to libp2p peers

binier · binier · commit b5c83e8fb627 · 2025-01-22T15:22:57.000+04:00
diff --git a/node/src/action_kind.rs b/node/src/action_kind.rs
@@ -425,6 +425,7 @@ pub enum ActionKind {
     P2pNetworkPubsubSignError,
     P2pNetworkPubsubValidateIncomingMessage,
     P2pNetworkPubsubValidateIncomingMessages,
+    P2pNetworkPubsubWebRtcRebroadcast,
     P2pNetworkPubsubEffectfulSign,
     P2pNetworkPubsubEffectfulValidateIncomingMessages,
     P2pNetworkRpcHeartbeatSend,
@@ -717,7 +718,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 607;
+    pub const COUNT: u16 = 608;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1961,6 +1962,7 @@ impl ActionKindGet for P2pNetworkPubsubAction {
             }
             Self::Graft { .. } => ActionKind::P2pNetworkPubsubGraft,
             Self::Prune { .. } => ActionKind::P2pNetworkPubsubPrune,
+            Self::WebRtcRebroadcast { .. } => ActionKind::P2pNetworkPubsubWebRtcRebroadcast,
             Self::Broadcast { .. } => ActionKind::P2pNetworkPubsubBroadcast,
             Self::Sign { .. } => ActionKind::P2pNetworkPubsubSign,
             Self::SignError { .. } => ActionKind::P2pNetworkPubsubSignError,
diff --git a/node/src/snark_pool/snark_pool_reducer.rs b/node/src/snark_pool/snark_pool_reducer.rs
@@ -202,15 +202,13 @@ impl SnarkPoolState {
                     }
                 }
 
-                // TODO: we only rebroadcast locally produced snarks here.
-                // libp2p logic already broadcasts everything right now and doesn't
+                // TODO: libp2p logic already broadcasts everything right now and doesn't
                 // wait for validation, thad needs to be fixed. See #952
-                if *is_sender_local {
-                    dispatcher.push(P2pChannelsSnarkAction::Libp2pBroadcast {
-                        snark: snark.clone(),
-                        nonce: 0,
-                    });
-                }
+                dispatcher.push(P2pChannelsSnarkAction::Libp2pBroadcast {
+                    snark: snark.clone(),
+                    nonce: 0,
+                    is_local: *is_sender_local,
+                });
             }
             SnarkPoolAction::P2pSendAll { .. } => {
                 let (dispatcher, global_state) = state_context.into_dispatcher_and_state();
diff --git a/node/src/transaction_pool/transaction_pool_actions.rs b/node/src/transaction_pool/transaction_pool_actions.rs
@@ -69,6 +69,7 @@ pub enum TransactionPoolAction {
     Rebroadcast {
         accepted: Vec<ValidCommandWithHash>,
         rejected: Vec<(ValidCommandWithHash, diff::Error)>,
+        is_local: bool,
     },
     CollectTransactionsByFee,
     #[action_event(level = trace)]
@@ -115,9 +116,9 @@ impl redux::EnablingCondition<crate::State> for TransactionPoolAction {
                         last_index,
                     )
                 }),
-            TransactionPoolAction::Rebroadcast { accepted, rejected } => {
-                !(accepted.is_empty() && rejected.is_empty())
-            }
+            TransactionPoolAction::Rebroadcast {
+                accepted, rejected, ..
+            } => !(accepted.is_empty() && rejected.is_empty()),
             _ => true,
         }
     }
diff --git a/node/src/transaction_pool/transaction_pool_reducer.rs b/node/src/transaction_pool/transaction_pool_reducer.rs
@@ -300,11 +300,14 @@ impl TransactionPoolState {
                 if let Some(rpc_action) = rpc_action {
                     dispatcher.push(rpc_action);
                 }
-                // TODO: we only rebroadcast locally injected transactions here.
-                // libp2p logic already broadcasts everything right now and doesn't
+                // TODO: libp2p logic already broadcasts everything right now and doesn't
                 // wait for validation, thad needs to be fixed. See #952
-                if is_sender_local && was_accepted {
-                    dispatcher.push(TransactionPoolAction::Rebroadcast { accepted, rejected });
+                if was_accepted {
+                    dispatcher.push(TransactionPoolAction::Rebroadcast {
+                        accepted,
+                        rejected,
+                        is_local: is_sender_local,
+                    });
                 }
             }
             TransactionPoolAction::ApplyTransitionFrontierDiff {
@@ -372,7 +375,11 @@ impl TransactionPoolState {
                     );
                 }
             }
-            TransactionPoolAction::Rebroadcast { accepted, rejected } => {
+            TransactionPoolAction::Rebroadcast {
+                accepted,
+                rejected,
+                is_local,
+            } => {
                 let rejected = rejected.iter().map(|(cmd, _)| cmd.data.forget_check());
 
                 let all_commands = accepted
@@ -387,6 +394,7 @@ impl TransactionPoolState {
                     dispatcher.push(P2pChannelsTransactionAction::Libp2pBroadcast {
                         transaction: Box::new((&cmd).into()),
                         nonce: 0,
+                        is_local: *is_local,
                     });
                 }
             }
diff --git a/node/src/transition_frontier/transition_frontier_effects.rs b/node/src/transition_frontier/transition_frontier_effects.rs
@@ -1,9 +1,11 @@
+use mina_p2p_messages::gossip::GossipNetMessageV2;
 use redux::Timestamp;
 
 use crate::block_producer::BlockProducerAction;
 use crate::consensus::ConsensusAction;
 use crate::ledger::LEDGER_DEPTH;
 use crate::p2p::channels::best_tip::P2pChannelsBestTipAction;
+use crate::p2p::P2pNetworkPubsubAction;
 use crate::snark_pool::{SnarkPoolAction, SnarkWork};
 use crate::stats::sync::SyncingLedger;
 use crate::{Store, TransactionPoolAction};
@@ -305,6 +307,9 @@ fn synced_effects<S: crate::Service>(
             best_tip: best_tip.block.clone(),
         });
     }
+    store.dispatch(P2pNetworkPubsubAction::WebRtcRebroadcast {
+        message: GossipNetMessageV2::NewState(best_tip.block().clone()),
+    });
 
     let best_tip_hash = best_tip.merkle_root_hash().clone();
     store.dispatch(ConsensusAction::Prune);
diff --git a/p2p/src/channels/snark/p2p_channels_snark_actions.rs b/p2p/src/channels/snark/p2p_channels_snark_actions.rs
@@ -54,6 +54,7 @@ pub enum P2pChannelsSnarkAction {
     Libp2pBroadcast {
         snark: Snark,
         nonce: u32,
+        is_local: bool,
     },
 }
 
diff --git a/p2p/src/channels/snark/p2p_channels_snark_reducer.rs b/p2p/src/channels/snark/p2p_channels_snark_reducer.rs
@@ -210,18 +210,26 @@ impl P2pChannelsSnarkState {
                 }
                 Ok(())
             }
+            #[cfg(not(feature = "p2p-libp2p"))]
+            P2pChannelsSnarkAction::Libp2pBroadcast { .. } => Ok(()),
             #[cfg(feature = "p2p-libp2p")]
-            P2pChannelsSnarkAction::Libp2pBroadcast { snark, nonce } => {
+            P2pChannelsSnarkAction::Libp2pBroadcast {
+                snark,
+                nonce,
+                is_local,
+            } => {
                 let dispatcher = state_context.into_dispatcher();
                 let message = Box::new((snark.statement(), (&snark).into()));
                 let message = v2::NetworkPoolSnarkPoolDiffVersionedStableV2::AddSolvedWork(message);
                 let nonce = nonce.into();
                 let message = GossipNetMessageV2::SnarkPoolDiff { message, nonce };
-                dispatcher.push(P2pNetworkPubsubAction::Broadcast { message });
+                if is_local {
+                    dispatcher.push(P2pNetworkPubsubAction::Broadcast { message });
+                } else {
+                    dispatcher.push(P2pNetworkPubsubAction::WebRtcRebroadcast { message });
+                }
                 Ok(())
             }
-            #[cfg(not(feature = "p2p-libp2p"))]
-            P2pChannelsSnarkAction::Libp2pBroadcast { .. } => Ok(()),
             P2pChannelsSnarkAction::Libp2pReceived { peer_id, snark, .. } => {
                 let (dispatcher, state) = state_context.into_dispatcher_and_state();
                 let p2p_state: &P2pState = state.substate()?;
diff --git a/p2p/src/channels/transaction/p2p_channels_transaction_actions.rs b/p2p/src/channels/transaction/p2p_channels_transaction_actions.rs
@@ -52,6 +52,7 @@ pub enum P2pChannelsTransactionAction {
     Libp2pBroadcast {
         transaction: Box<Transaction>,
         nonce: u32,
+        is_local: bool,
     },
 }
 
diff --git a/p2p/src/channels/transaction/p2p_channels_transaction_reducer.rs b/p2p/src/channels/transaction/p2p_channels_transaction_reducer.rs
@@ -230,14 +230,22 @@ impl P2pChannelsTransactionState {
             #[cfg(not(feature = "p2p-libp2p"))]
             P2pChannelsTransactionAction::Libp2pBroadcast { .. } => Ok(()),
             #[cfg(feature = "p2p-libp2p")]
-            P2pChannelsTransactionAction::Libp2pBroadcast { transaction, nonce } => {
+            P2pChannelsTransactionAction::Libp2pBroadcast {
+                transaction,
+                nonce,
+                is_local,
+            } => {
                 let dispatcher = state_context.into_dispatcher();
                 let message = v2::NetworkPoolTransactionPoolDiffVersionedStableV2(
                     std::iter::once(*transaction).collect(),
                 );
                 let nonce = nonce.into();
                 let message = GossipNetMessageV2::TransactionPoolDiff { message, nonce };
-                dispatcher.push(P2pNetworkPubsubAction::Broadcast { message });
+                if is_local {
+                    dispatcher.push(P2pNetworkPubsubAction::Broadcast { message });
+                } else {
+                    dispatcher.push(P2pNetworkPubsubAction::WebRtcRebroadcast { message });
+                }
                 Ok(())
             }
         }
diff --git a/p2p/src/identity/secret_key.rs b/p2p/src/identity/secret_key.rs
@@ -178,6 +178,15 @@ impl SecretKey {
     pub fn libp2p_pubsub_sign(&mut self, msg: &[u8]) -> Signature {
         self.sign(&[b"libp2p-pubsub:", msg].concat())
     }
+
+    pub fn libp2p_pubsub_pb_message_sign(
+        &mut self,
+        msg: &crate::pb::Message,
+    ) -> Result<Signature, prost::EncodeError> {
+        let mut buf = Vec::new();
+        prost::Message::encode(msg, &mut buf)?;
+        Ok(self.libp2p_pubsub_sign(&buf))
+    }
 }
 
 pub trait EncryptableType: Serialize + for<'a> Deserialize<'a> {
diff --git a/p2p/src/network/pubsub/mod.rs b/p2p/src/network/pubsub/mod.rs
@@ -20,6 +20,13 @@ const TOPIC: &str = "coda/consensus-messages/0.0.1";
 pub mod pubsub_effectful;
 pub use pubsub_effectful::P2pNetworkPubsubEffectfulAction;
 
+use binprot::BinProtWrite;
+use mina_p2p_messages::gossip::GossipNetMessageV2;
+use openmina_core::bug_condition;
+use sha2::{Digest, Sha256};
+
+use crate::identity::SecretKey;
+
 #[derive(serde::Serialize, serde:: Deserialize, Debug, Clone)]
 pub enum BroadcastMessageId {
     BlockHash {
@@ -29,3 +36,29 @@ pub enum BroadcastMessageId {
         message_id: P2pNetworkPubsubMessageCacheId,
     },
 }
+
+pub(super) fn webrtc_source_sk(message: &GossipNetMessageV2) -> SecretKey {
+    let mut hasher = Sha256::new();
+    if let Err(err) = message.binprot_write(&mut hasher) {
+        bug_condition!("trying to broadcast message which can't be binprot serialized! err: {err}");
+        return SecretKey::from_bytes([0; 32]);
+    }
+    SecretKey::from_bytes(hasher.finalize().into())
+}
+
+pub(super) fn webrtc_source_sk_from_bytes(bytes: &[u8]) -> SecretKey {
+    let mut hasher = Sha256::new();
+    hasher.update(bytes);
+    SecretKey::from_bytes(hasher.finalize().into())
+}
+
+pub(super) fn encode_message(message: &GossipNetMessageV2) -> std::io::Result<Vec<u8>> {
+    let mut buffer = vec![0; 8];
+
+    message.binprot_write(&mut buffer)?;
+
+    let len = buffer.len() - 8;
+    buffer[..8].clone_from_slice(&(len as u64).to_le_bytes());
+
+    Ok(buffer)
+}
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_actions.rs b/p2p/src/network/pubsub/p2p_network_pubsub_actions.rs
@@ -77,6 +77,11 @@ pub enum P2pNetworkPubsubAction {
         topic_id: String,
     },
 
+    /// Rebroadcast message received from WebRTC connection.
+    WebRtcRebroadcast {
+        message: GossipNetMessageV2,
+    },
+
     /// Initiate the broadcasting of a message to all subscribed peers.
     ///
     /// **Fields:**
@@ -190,6 +195,17 @@ impl redux::EnablingCondition<P2pState> for P2pNetworkPubsubAction {
                 .topics
                 .get(topic_id)
                 .is_some_and(|topics| topics.contains_key(peer_id)),
+            P2pNetworkPubsubAction::WebRtcRebroadcast { message } => {
+                let source = super::webrtc_source_sk(message)
+                    .public_key()
+                    .peer_id()
+                    .try_into()
+                    .unwrap();
+                pubsub
+                    .mcache
+                    .get_message(&P2pNetworkPubsubMessageCacheId { source, seqno: 0 })
+                    .is_none()
+            }
             P2pNetworkPubsubAction::BroadcastValidatedMessage { message_id }
             | P2pNetworkPubsubAction::RejectMessage {
                 message_id: Some(message_id),
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs b/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs
@@ -389,18 +389,74 @@ impl P2pNetworkPubsubState {
                 Ok(())
             }
             P2pNetworkPubsubAction::OutgoingMessageError { .. } => Ok(()),
-            P2pNetworkPubsubAction::Broadcast { message } => {
-                let mut buffer = vec![0; 8];
+            P2pNetworkPubsubAction::WebRtcRebroadcast { message } => {
+                let data = match super::encode_message(&message) {
+                    Err(err) => {
+                        bug_condition!("binprot serialization error: {err}");
+                        return Ok(());
+                    }
+                    Ok(data) => data,
+                };
 
-                if binprot::BinProtWrite::binprot_write(&message, &mut buffer).is_err() {
-                    bug_condition!("binprot serialization error");
-                    return Ok(());
-                }
+                let mut source_sk = super::webrtc_source_sk_from_bytes(&data[8..]);
+                let source_peer_id = source_sk.public_key().peer_id();
+                let message_id = P2pNetworkPubsubMessageCacheId {
+                    source: libp2p_identity::PeerId::try_from(source_peer_id).unwrap(),
+                    seqno: 0,
+                };
+                let mut msg = pb::Message {
+                    from: Some(message_id.source.to_bytes().to_vec()),
+                    data: Some(data),
+                    seqno: Some(message_id.seqno.to_be_bytes().to_vec()),
+                    topic: super::TOPIC.to_owned(),
+                    signature: None,
+                    key: None,
+                };
 
-                let len = buffer.len() - 8;
-                buffer[..8].clone_from_slice(&(len as u64).to_le_bytes());
+                msg.signature = match source_sk.libp2p_pubsub_pb_message_sign(&msg) {
+                    Err(err) => {
+                        bug_condition!("pubsub prost encode error: {err}");
+                        return Ok(());
+                    }
+                    Ok(sig) => Some(sig.to_bytes().to_vec()),
+                };
 
-                Self::prepare_to_sign(state_context, buffer)
+                let message_state = match &message {
+                    GossipNetMessageV2::NewState(block) => {
+                        P2pNetworkPubsubMessageCacheMessage::PreValidatedBlockMessage {
+                            block_hash: block.try_hash()?,
+                            message: msg,
+                            peer_id: source_peer_id,
+                            time,
+                        }
+                    }
+                    _ => P2pNetworkPubsubMessageCacheMessage::PreValidated {
+                        message: msg,
+                        peer_id: source_peer_id,
+                        time,
+                    },
+                };
+
+                pubsub_state.mcache.map.insert(message_id, message_state);
+
+                let dispatcher = state_context.into_dispatcher();
+
+                dispatcher.push(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+                    message_id: super::BroadcastMessageId::MessageId { message_id },
+                });
+
+                Ok(())
+            }
+            P2pNetworkPubsubAction::Broadcast { message } => {
+                let data = match super::encode_message(&message) {
+                    Err(err) => {
+                        bug_condition!("binprot serialization error: {err}");
+                        return Ok(());
+                    }
+                    Ok(data) => data,
+                };
+
+                Self::prepare_to_sign(state_context, data)
             }
             P2pNetworkPubsubAction::Sign {
                 seqno,
@@ -428,11 +484,7 @@ impl P2pNetworkPubsubState {
                 };
 
                 let dispatcher = state_context.into_dispatcher();
-                dispatcher.push(P2pNetworkPubsubEffectfulAction::Sign {
-                    author,
-                    topic,
-                    message,
-                });
+                dispatcher.push(P2pNetworkPubsubEffectfulAction::Sign { author, message });
                 Ok(())
             }
             P2pNetworkPubsubAction::SignError { .. } => {
diff --git a/p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_actions.rs b/p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_actions.rs
diff --git a/p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_effects.rs b/p2p/src/network/pubsub/pubsub_effectful/p2p_network_pubsub_effectful_effects.rs

Original file line number	Diff line number	Diff line change
`@@ -300,11 +300,14 @@ impl TransactionPoolState {`
`300`	`300`	`if let Some(rpc_action) = rpc_action {`
`301`	`301`	`dispatcher.push(rpc_action);`
`302`	`302`	`}`
`303`		`- // TODO: we only rebroadcast locally injected transactions here.`
`304`		`- // libp2p logic already broadcasts everything right now and doesn't`
	`303`	`+ // TODO: libp2p logic already broadcasts everything right now and doesn't`
`305`	`304`	`// wait for validation, thad needs to be fixed. See #952`
`306`		`- if is_sender_local && was_accepted {`
`307`		`- dispatcher.push(TransactionPoolAction::Rebroadcast { accepted, rejected });`
	`305`	`+ if was_accepted {`
	`306`	`+ dispatcher.push(TransactionPoolAction::Rebroadcast {`
	`307`	`+ accepted,`
	`308`	`+ rejected,`
	`309`	`+ is_local: is_sender_local,`
	`310`	`+ });`
`308`	`311`	`}`
`309`	`312`	`}`
`310`	`313`	`TransactionPoolAction::ApplyTransitionFrontierDiff {`
`@@ -372,7 +375,11 @@ impl TransactionPoolState {`
`372`	`375`	`);`
`373`	`376`	`}`
`374`	`377`	`}`
`375`		`- TransactionPoolAction::Rebroadcast { accepted, rejected } => {`
	`378`	`+ TransactionPoolAction::Rebroadcast {`
	`379`	`+ accepted,`
	`380`	`+ rejected,`
	`381`	`+ is_local,`
	`382`	`+ } => {`
`376`	`383`	`let rejected = rejected.iter().map(\|(cmd, _)\| cmd.data.forget_check());`
`377`	`384`
`378`	`385`	`let all_commands = accepted`
`@@ -387,6 +394,7 @@ impl TransactionPoolState {`
`387`	`394`	`dispatcher.push(P2pChannelsTransactionAction::Libp2pBroadcast {`
`388`	`395`	`transaction: Box::new((&cmd).into()),`
`389`	`396`	`nonce: 0,`
	`397`	`+ is_local: *is_local,`
`390`	`398`	`});`
`391`	`399`	`}`
`392`	`400`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,6 +54,7 @@ pub enum P2pChannelsSnarkAction {`
`54`	`54`	`Libp2pBroadcast {`
`55`	`55`	`snark: Snark,`
`56`	`56`	`nonce: u32,`
	`57`	`+ is_local: bool,`
`57`	`58`	`},`
`58`	`59`	`}`
`59`	`60`
Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,7 @@ pub enum P2pChannelsTransactionAction {`
`52`	`52`	`Libp2pBroadcast {`
`53`	`53`	`transaction: Box<Transaction>,`
`54`	`54`	`nonce: u32,`
	`55`	`+ is_local: bool,`
`55`	`56`	`},`
`56`	`57`	`}`
`57`	`58`