Fix WebRTC to LibP2P propagation

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(

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,

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();

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,
         }
     }

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,
                     });
                 }
             }

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.

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);

p2p/src/channels/snark/p2p_channels_snark_actions.rs

@@ -54,6 +54,7 @@ pub enum P2pChannelsSnarkAction {
     Libp2pBroadcast {
         snark: Snark,
         nonce: u32,
+        is_local: bool,
     },
 }
 

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()?;

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,
     },
 }
 

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(())
             }
         }

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;

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> {