Merge pull request #1033 from openmina/fix/p2p/webrtc_to_libp2p_propagation

binier · web-flow · commit 2676e39b59db · 2025-01-23T12:27:50.000+04:00
Fix WebRTC to LibP2P propagation
diff --git a/node/common/src/service/p2p.rs b/node/common/src/service/p2p.rs
@@ -123,28 +123,23 @@ impl P2pCryptoService for NodeService {
     fn sign_key(&mut self, key: &[u8; 32]) -> Vec<u8> {
         // TODO: make deterministic
         let msg = [b"noise-libp2p-static-key:", key.as_slice()].concat();
-        let sig = self
-            .p2p
-            .mio
-            .keypair()
-            .sign(&msg)
-            .expect("unable to create signature");
+        let sig = self.p2p.sec_key.sign(&msg);
+        let libp2p_sec_key = libp2p_identity::Keypair::try_from(self.p2p.sec_key.clone()).unwrap();
 
         let mut payload = vec![];
         payload.extend_from_slice(b"\x0a\x24");
-        payload.extend_from_slice(&self.p2p.mio.keypair().public().encode_protobuf());
+        payload.extend_from_slice(&libp2p_sec_key.public().encode_protobuf());
         payload.extend_from_slice(b"\x12\x40");
-        payload.extend_from_slice(&sig);
+        payload.extend_from_slice(&sig.to_bytes());
         payload
     }
 
     fn sign_publication(&mut self, publication: &[u8]) -> Vec<u8> {
-        let msg = [b"libp2p-pubsub:", publication].concat();
         self.p2p
-            .mio
-            .keypair()
-            .sign(&msg)
-            .expect("unable to create signature")
+            .sec_key
+            .libp2p_pubsub_sign(publication)
+            .to_bytes()
+            .to_vec()
     }
 
     fn verify_publication(
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/sync/transition_frontier_sync_effects.rs b/node/src/transition_frontier/sync/transition_frontier_sync_effects.rs
@@ -322,11 +322,6 @@ impl TransitionFrontierSyncAction {
                 if !store.dispatch(TransitionFrontierSyncAction::BlocksNextApplyInit) {
                     store.dispatch(TransitionFrontierSyncAction::BlocksSuccess);
                 }
-
-                // TODO this should be handled by a callback
-                store.dispatch(P2pNetworkPubsubAction::BroadcastValidatedMessage {
-                    message_id: p2p::BroadcastMessageId::BlockHash { hash: hash.clone() },
-                });
             }
             TransitionFrontierSyncAction::BlocksSuccess => {}
             // Bootstrap/Catchup is practically complete at this point.
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,18 @@ fn synced_effects<S: crate::Service>(
             best_tip: best_tip.block.clone(),
         });
     }
+    // TODO this should be handled by a callback
+    // If this get dispatched, we received block from libp2p.
+    if !store.dispatch(P2pNetworkPubsubAction::BroadcastValidatedMessage {
+        message_id: p2p::BroadcastMessageId::BlockHash {
+            hash: best_tip.hash().clone(),
+        },
+    }) {
+        // Otherwise block was received from WebRTC so inject it in libp2p.
+        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,27 @@ 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 {
+                    // rebroadcast snark if received from webrtc network, otherwise noop.
+                    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,23 @@ 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 {
+                    // rebroadcast block if received from webrtc network, otherwise noop.
+                    dispatcher.push(P2pNetworkPubsubAction::WebRtcRebroadcast { message });
+                }
                 Ok(())
             }
         }
diff --git a/p2p/src/identity/mod.rs b/p2p/src/identity/mod.rs
@@ -6,3 +6,6 @@ pub use public_key::PublicKey;
 
 mod secret_key;
 pub use secret_key::{EncryptableType, SecretKey};
+
+mod signature;
+pub use signature::Signature;
diff --git a/p2p/src/identity/secret_key.rs b/p2p/src/identity/secret_key.rs
@@ -1,12 +1,12 @@
 use std::{fmt, path::Path, str::FromStr};
 
 use base64::Engine;
-use ed25519_dalek::SigningKey as Ed25519SecretKey;
+use ed25519_dalek::{ed25519::signature::SignerMut, SigningKey as Ed25519SecretKey};
 use openmina_core::{EncryptedSecretKey, EncryptedSecretKeyFile, EncryptionError};
 use rand::{CryptoRng, Rng};
 use serde::{Deserialize, Serialize};
 
-use crate::identity::PublicKey;
+use super::{PublicKey, Signature};
 
 #[derive(Clone)]
 pub struct SecretKey(Ed25519SecretKey);
@@ -170,6 +170,23 @@ impl SecretKey {
         let data: Vec<u8> = self.decrypt_raw(other_pk, ciphertext.as_ref())?;
         serde_json::from_slice(&data).map_err(Box::<dyn std::error::Error>::from)
     }
+
+    pub fn sign(&mut self, data: &[u8]) -> Signature {
+        Signature(self.0.sign(data))
+    }
+
+    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/identity/signature.rs b/p2p/src/identity/signature.rs
diff --git a/p2p/src/network/pubsub/mod.rs b/p2p/src/network/pubsub/mod.rs
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_actions.rs b/p2p/src/network/pubsub/p2p_network_pubsub_actions.rs
diff --git a/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs b/p2p/src/network/pubsub/p2p_network_pubsub_reducer.rs
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
`@@ -322,11 +322,6 @@ impl TransitionFrontierSyncAction {`
`322`	`322`	`if !store.dispatch(TransitionFrontierSyncAction::BlocksNextApplyInit) {`
`323`	`323`	`store.dispatch(TransitionFrontierSyncAction::BlocksSuccess);`
`324`	`324`	`}`
`325`		`-`
`326`		`- // TODO this should be handled by a callback`
`327`		`- store.dispatch(P2pNetworkPubsubAction::BroadcastValidatedMessage {`
`328`		`- message_id: p2p::BroadcastMessageId::BlockHash { hash: hash.clone() },`
`329`		`- });`
`330`	`325`	`}`
`331`	`326`	`TransitionFrontierSyncAction::BlocksSuccess => {}`
`332`	`327`	`// Bootstrap/Catchup is practically complete at this point.`
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`