feat(p2p/webrtc): implement channel for exchanging relayed signaling messages

re: #772

Author: binier

node/common/src/service/p2p.rs

@@ -3,7 +3,11 @@ use std::collections::BTreeMap;
 use node::{
     core::channels::mpsc,
     event_source::Event,
-    p2p::{connection::outgoing::P2pConnectionOutgoingInitOpts, PeerId},
+    p2p::{
+        connection::outgoing::P2pConnectionOutgoingInitOpts,
+        identity::{EncryptableType, PublicKey},
+        PeerId,
+    },
 };
 use rand::prelude::*;
 #[cfg(feature = "p2p-libp2p")]
@@ -34,6 +38,23 @@ impl webrtc::P2pServiceWebrtc for NodeService {
     fn peers(&mut self) -> &mut BTreeMap<PeerId, webrtc::PeerState> {
         &mut self.p2p.webrtc.peers
     }
+
+    fn encrypt<T: EncryptableType>(
+        &mut self,
+        other_pk: &PublicKey,
+        message: &T,
+    ) -> Result<T::Encrypted, ()> {
+        let rng = &mut self.rng;
+        self.p2p.sec_key.encrypt(other_pk, rng, message)
+    }
+
+    fn decrypt<T: EncryptableType>(
+        &mut self,
+        other_pk: &PublicKey,
+        encrypted: &T::Encrypted,
+    ) -> Result<T, ()> {
+        self.p2p.sec_key.decrypt(other_pk, encrypted)
+    }
 }
 
 impl webrtc_with_libp2p::P2pServiceWebrtcWithLibp2p for NodeService {

node/src/action_kind.rs

@@ -28,6 +28,8 @@ use crate::p2p::channels::best_tip::P2pChannelsBestTipAction;
 use crate::p2p::channels::best_tip_effectful::P2pChannelsBestTipEffectfulAction;
 use crate::p2p::channels::rpc::P2pChannelsRpcAction;
 use crate::p2p::channels::rpc_effectful::P2pChannelsRpcEffectfulAction;
+use crate::p2p::channels::signaling::exchange::P2pChannelsSignalingExchangeAction;
+use crate::p2p::channels::signaling::exchange_effectful::P2pChannelsSignalingExchangeEffectfulAction;
 use crate::p2p::channels::snark::P2pChannelsSnarkAction;
 use crate::p2p::channels::snark_effectful::P2pChannelsSnarkEffectfulAction;
 use crate::p2p::channels::snark_job_commitment::P2pChannelsSnarkJobCommitmentAction;
@@ -209,6 +211,21 @@ pub enum ActionKind {
     P2pChannelsRpcEffectfulInit,
     P2pChannelsRpcEffectfulRequestSend,
     P2pChannelsRpcEffectfulResponseSend,
+    P2pChannelsSignalingExchangeAnswerReceived,
+    P2pChannelsSignalingExchangeAnswerSend,
+    P2pChannelsSignalingExchangeInit,
+    P2pChannelsSignalingExchangeOfferDecryptError,
+    P2pChannelsSignalingExchangeOfferDecryptSuccess,
+    P2pChannelsSignalingExchangeOfferReceived,
+    P2pChannelsSignalingExchangeOfferSend,
+    P2pChannelsSignalingExchangePending,
+    P2pChannelsSignalingExchangeReady,
+    P2pChannelsSignalingExchangeRequestReceived,
+    P2pChannelsSignalingExchangeRequestSend,
+    P2pChannelsSignalingExchangeEffectfulAnswerEncryptAndSend,
+    P2pChannelsSignalingExchangeEffectfulInit,
+    P2pChannelsSignalingExchangeEffectfulMessageSend,
+    P2pChannelsSignalingExchangeEffectfulOfferDecrypt,
     P2pChannelsSnarkInit,
     P2pChannelsSnarkLibp2pBroadcast,
     P2pChannelsSnarkLibp2pReceived,
@@ -618,7 +635,7 @@ pub enum ActionKind {
 }
 
 impl ActionKind {
-    pub const COUNT: u16 = 510;
+    pub const COUNT: u16 = 525;
 }
 
 impl std::fmt::Display for ActionKind {
@@ -1029,6 +1046,7 @@ impl ActionKindGet for P2pChannelsAction {
     fn kind(&self) -> ActionKind {
         match self {
             Self::MessageReceived(a) => a.kind(),
+            Self::SignalingExchange(a) => a.kind(),
             Self::BestTip(a) => a.kind(),
             Self::Transaction(a) => a.kind(),
             Self::Snark(a) => a.kind(),
@@ -1069,6 +1087,7 @@ impl ActionKindGet for P2pNetworkAction {
 impl ActionKindGet for P2pChannelsEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {
+            Self::SignalingExchange(a) => a.kind(),
             Self::BestTip(a) => a.kind(),
             Self::Rpc(a) => a.kind(),
             Self::Snark(a) => a.kind(),
@@ -1400,6 +1419,28 @@ impl ActionKindGet for P2pChannelsMessageReceivedAction {
     }
 }
 
+impl ActionKindGet for P2pChannelsSignalingExchangeAction {
+    fn kind(&self) -> ActionKind {
+        match self {
+            Self::Init { .. } => ActionKind::P2pChannelsSignalingExchangeInit,
+            Self::Pending { .. } => ActionKind::P2pChannelsSignalingExchangePending,
+            Self::Ready { .. } => ActionKind::P2pChannelsSignalingExchangeReady,
+            Self::RequestSend { .. } => ActionKind::P2pChannelsSignalingExchangeRequestSend,
+            Self::OfferReceived { .. } => ActionKind::P2pChannelsSignalingExchangeOfferReceived,
+            Self::OfferDecryptError { .. } => {
+                ActionKind::P2pChannelsSignalingExchangeOfferDecryptError
+            }
+            Self::OfferDecryptSuccess { .. } => {
+                ActionKind::P2pChannelsSignalingExchangeOfferDecryptSuccess
+            }
+            Self::AnswerSend { .. } => ActionKind::P2pChannelsSignalingExchangeAnswerSend,
+            Self::RequestReceived { .. } => ActionKind::P2pChannelsSignalingExchangeRequestReceived,
+            Self::OfferSend { .. } => ActionKind::P2pChannelsSignalingExchangeOfferSend,
+            Self::AnswerReceived { .. } => ActionKind::P2pChannelsSignalingExchangeAnswerReceived,
+        }
+    }
+}
+
 impl ActionKindGet for P2pChannelsBestTipAction {
     fn kind(&self) -> ActionKind {
         match self {
@@ -1652,6 +1693,23 @@ impl ActionKindGet for P2pNetworkRpcAction {
     }
 }
 
+impl ActionKindGet for P2pChannelsSignalingExchangeEffectfulAction {
+    fn kind(&self) -> ActionKind {
+        match self {
+            Self::Init { .. } => ActionKind::P2pChannelsSignalingExchangeEffectfulInit,
+            Self::MessageSend { .. } => {
+                ActionKind::P2pChannelsSignalingExchangeEffectfulMessageSend
+            }
+            Self::OfferDecrypt { .. } => {
+                ActionKind::P2pChannelsSignalingExchangeEffectfulOfferDecrypt
+            }
+            Self::AnswerEncryptAndSend { .. } => {
+                ActionKind::P2pChannelsSignalingExchangeEffectfulAnswerEncryptAndSend
+            }
+        }
+    }
+}
+
 impl ActionKindGet for P2pChannelsBestTipEffectfulAction {
     fn kind(&self) -> ActionKind {
         match self {

node/src/event_source/event_source_effects.rs

@@ -1,3 +1,4 @@
+use p2p::channels::signaling::exchange::P2pChannelsSignalingExchangeAction;
 use p2p::channels::snark::P2pChannelsSnarkAction;
 use p2p::channels::streaming_rpc::P2pChannelsStreamingRpcAction;
 use p2p::channels::transaction::P2pChannelsTransactionAction;
@@ -200,8 +201,12 @@ pub fn event_source_effects<S: Service>(store: &mut Store<S>, action: EventSourc
                             openmina_core::log::warn!(meta.time(); kind = "P2pChannelEvent::Opened", peer_id = peer_id.to_string(), error = err);
                             // TODO(binier): dispatch error action.
                         }
-                        // TODO(binier): maybe dispatch success and then ready.
                         Ok(_) => match chan_id {
+                            ChannelId::SignalingExchange => {
+                                store.dispatch(P2pChannelsSignalingExchangeAction::Ready {
+                                    peer_id,
+                                });
+                            }
                             ChannelId::BestTipPropagation => {
                                 store.dispatch(P2pChannelsBestTipAction::Ready { peer_id });
                             }

node/src/logger/logger_effects.rs

@@ -69,6 +69,7 @@ pub fn logger_effects<S: Service>(store: &Store<S>, action: ActionWithMetaRef<'_
             P2pAction::Identify(action) => action.action_event(&context),
             P2pAction::Channels(action) => match action {
                 P2pChannelsAction::MessageReceived(action) => action.action_event(&context),
+                P2pChannelsAction::SignalingExchange(action) => action.action_event(&context),
                 P2pChannelsAction::BestTip(action) => action.action_event(&context),
                 P2pChannelsAction::Transaction(action) => action.action_event(&context),
                 P2pChannelsAction::Snark(action) => action.action_event(&context),
@@ -105,6 +106,9 @@ pub fn logger_effects<S: Service>(store: &Store<S>, action: ActionWithMetaRef<'_
         },
         Action::P2pEffectful(action) => match action {
             p2p::P2pEffectfulAction::Channels(action) => match action {
+                P2pChannelsEffectfulAction::SignalingExchange(action) => {
+                    action.action_event(&context)
+                }
                 P2pChannelsEffectfulAction::BestTip(action) => action.action_event(&context),
                 P2pChannelsEffectfulAction::Rpc(action) => action.action_event(&context),
                 P2pChannelsEffectfulAction::StreamingRpc(action) => action.action_event(&context),

node/src/p2p/channels/mod.rs

@@ -2,6 +2,7 @@ pub use ::p2p::channels::*;
 
 pub mod best_tip;
 pub mod rpc;
+pub mod signaling;
 pub mod snark;
 pub mod snark_job_commitment;
 pub mod streaming_rpc;

node/src/p2p/channels/signaling/mod.rs

@@ -0,0 +1,3 @@
+pub use ::p2p::channels::signaling::*;
+
+mod p2p_channels_signaling_exchange_actions;

node/src/p2p/channels/signaling/p2p_channels_signaling_exchange_actions.rs

@@ -0,0 +1,7 @@
+use super::exchange::*;
+
+impl redux::EnablingCondition<crate::State> for P2pChannelsSignalingExchangeAction {
+    fn is_enabled(&self, state: &crate::State, time: redux::Timestamp) -> bool {
+        state.p2p.is_enabled(self, time)
+    }
+}

node/src/p2p/mod.rs

@@ -3,6 +3,7 @@ use p2p::{
     channels::{
         best_tip_effectful::P2pChannelsBestTipEffectfulAction,
         rpc_effectful::P2pChannelsRpcEffectfulAction,
+        signaling::exchange_effectful::P2pChannelsSignalingExchangeEffectfulAction,
         snark_effectful::P2pChannelsSnarkEffectfulAction,
         snark_job_commitment_effectful::P2pChannelsSnarkJobCommitmentEffectfulAction,
         streaming_rpc_effectful::P2pChannelsStreamingRpcEffectfulAction,
@@ -107,6 +108,7 @@ impl_into_global_action!(network::kad::P2pNetworkKademliaAction);
 impl_into_global_action!(network::pubsub::P2pNetworkPubsubAction);
 
 impl_into_global_action!(channels::P2pChannelsMessageReceivedAction);
+impl_into_global_action!(channels::signaling::exchange::P2pChannelsSignalingExchangeAction);
 impl_into_global_action!(channels::best_tip::P2pChannelsBestTipAction);
 impl_into_global_action!(channels::transaction::P2pChannelsTransactionAction);
 impl_into_global_action!(channels::snark::P2pChannelsSnarkAction);
@@ -132,6 +134,7 @@ impl_into_global_action!(effectful p2p::P2pNetworkPnetEffectfulAction);
 impl_into_global_action!(effectful connection::incoming_effectful::P2pConnectionIncomingEffectfulAction);
 impl_into_global_action!(effectful connection::outgoing_effectful::P2pConnectionOutgoingEffectfulAction);
 impl_into_global_action!(effectful p2p::disconnection_effectful::P2pDisconnectionEffectfulAction);
+impl_into_global_action!(effectful P2pChannelsSignalingExchangeEffectfulAction);
 impl_into_global_action!(effectful P2pChannelsBestTipEffectfulAction);
 impl_into_global_action!(effectful P2pChannelsStreamingRpcEffectfulAction);
 impl_into_global_action!(effectful P2pChannelsTransactionEffectfulAction);

node/testing/src/service/mod.rs

@@ -359,6 +359,22 @@ impl P2pServiceWebrtc for NodeTestingService {
     fn incoming_init(&mut self, peer_id: PeerId, offer: webrtc::Offer) {
         P2pServiceWebrtc::incoming_init(&mut self.real, peer_id, offer)
     }
+
+    fn encrypt<T: node::p2p::identity::EncryptableType>(
+        &mut self,
+        other_pk: &node::p2p::identity::PublicKey,
+        message: &T,
+    ) -> Result<T::Encrypted, ()> {
+        self.real.encrypt(other_pk, message)
+    }
+
+    fn decrypt<T: node::p2p::identity::EncryptableType>(
+        &mut self,
+        other_pub_key: &node::p2p::identity::PublicKey,
+        encrypted: &T::Encrypted,
+    ) -> Result<T, ()> {
+        self.real.decrypt(other_pub_key, encrypted)
+    }
 }
 
 impl P2pServiceWebrtcWithLibp2p for NodeTestingService {

p2p/src/channels/mod.rs

@@ -2,6 +2,7 @@ pub mod best_tip;
 pub mod best_tip_effectful;
 pub mod rpc;
 pub mod rpc_effectful;
+pub mod signaling;
 pub mod snark;
 pub mod snark_effectful;
 pub mod snark_job_commitment;
@@ -26,6 +27,7 @@ use binprot::{BinProtRead, BinProtWrite};
 use binprot_derive::{BinProtRead, BinProtWrite};
 use derive_more::From;
 use serde::{Deserialize, Serialize};
+use signaling::exchange::SignalingExchangeChannelMsg;
 use strum_macros::EnumIter;
 
 use self::best_tip::BestTipPropagationChannelMsg;
@@ -38,12 +40,13 @@ use self::transaction::TransactionPropagationChannelMsg;
 #[derive(Serialize, Deserialize, EnumIter, Debug, Ord, PartialOrd, Eq, PartialEq, Clone, Copy)]
 #[repr(u8)]
 pub enum ChannelId {
+    SignalingExchange = 1,
     BestTipPropagation = 2,
     TransactionPropagation = 3,
     SnarkPropagation = 4,
     SnarkJobCommitmentPropagation = 5,
-    Rpc = 100,
-    StreamingRpc = 101,
+    Rpc = 6,
+    StreamingRpc = 7,
 }
 
 impl ChannelId {
@@ -59,6 +62,7 @@ impl ChannelId {
 
     pub fn name(self) -> &'static str {
         match self {
+            Self::SignalingExchange => "signaling/exchange",
             Self::BestTipPropagation => "best_tip/propagation",
             Self::TransactionPropagation => "transaction/propagation",
             Self::SnarkPropagation => "snark/propagation",
@@ -70,6 +74,7 @@ impl ChannelId {
 
     pub fn supported_by_libp2p(self) -> bool {
         match self {
+            Self::SignalingExchange => false,
             Self::BestTipPropagation => true,
             Self::TransactionPropagation => true,
             Self::SnarkPropagation => true,
@@ -81,6 +86,8 @@ impl ChannelId {
 
     pub fn max_msg_size(self) -> usize {
         match self {
+            // TODO(binier): measure signaling message sizes
+            Self::SignalingExchange => 16 * 1024, // 16KB
             // TODO(binier): reduce this value once we change message for best tip
             // propagation to just propagating consensus state with block hash.
             Self::BestTipPropagation => 32 * 1024 * 1024, // 32MB
@@ -122,6 +129,7 @@ impl MsgId {
 
 #[derive(BinProtWrite, BinProtRead, Serialize, Deserialize, From, Debug, Clone)]
 pub enum ChannelMsg {
+    SignalingExchange(SignalingExchangeChannelMsg),
     BestTipPropagation(BestTipPropagationChannelMsg),
     TransactionPropagation(TransactionPropagationChannelMsg),
     SnarkPropagation(SnarkPropagationChannelMsg),
@@ -133,6 +141,7 @@ pub enum ChannelMsg {
 impl ChannelMsg {
     pub fn channel_id(&self) -> ChannelId {
         match self {
+            Self::SignalingExchange(_) => ChannelId::SignalingExchange,
             Self::BestTipPropagation(_) => ChannelId::BestTipPropagation,
             Self::TransactionPropagation(_) => ChannelId::TransactionPropagation,
             Self::SnarkPropagation(_) => ChannelId::SnarkPropagation,
@@ -147,6 +156,7 @@ impl ChannelMsg {
         W: std::io::Write,
     {
         match self {
+            Self::SignalingExchange(v) => v.binprot_write(w),
             Self::BestTipPropagation(v) => v.binprot_write(w),
             Self::TransactionPropagation(v) => v.binprot_write(w),
             Self::SnarkPropagation(v) => v.binprot_write(w),
@@ -162,6 +172,9 @@ impl ChannelMsg {
         R: std::io::Read + ?Sized,
     {
         match id {
+            ChannelId::SignalingExchange => {
+                SignalingExchangeChannelMsg::binprot_read(r).map(|v| v.into())
+            }
             ChannelId::BestTipPropagation => {
                 BestTipPropagationChannelMsg::binprot_read(r).map(|v| v.into())
             }
@@ -194,6 +207,11 @@ impl crate::P2pState {
         // exhaustive matching so that we don't miss any channels.
         for id in self.config.enabled_channels.iter().copied() {
             match id {
+                ChannelId::SignalingExchange => {
+                    dispatcher.push(
+                        signaling::exchange::P2pChannelsSignalingExchangeAction::Init { peer_id },
+                    );
+                }
                 ChannelId::BestTipPropagation => {
                     dispatcher.push(best_tip::P2pChannelsBestTipAction::Init { peer_id });
                 }