Merge pull request #846 from openmina/fix/p2p/incoming

Fix incoming peer handling

Author: vlad9486

node/src/rpc/mod.rs

@@ -24,6 +24,7 @@ mod rpc_actions;
 pub use rpc_actions::*;
 
 mod rpc_reducer;
+pub use rpc_reducer::collect_rpc_peers_info;
 
 mod rpc_impls;
 
@@ -164,6 +165,7 @@ pub struct RpcPeerInfo {
     pub best_tip_timestamp: Option<u64>,
     pub connection_status: PeerConnectionStatus,
     pub address: Option<String>,
+    pub incoming: bool,
     pub time: u64,
 }
 

node/src/rpc/rpc_reducer.rs

@@ -613,35 +613,38 @@ impl RpcState {
     }
 }
 
-fn collect_rpc_peers_info(state: &crate::State) -> Vec<RpcPeerInfo> {
+pub fn collect_rpc_peers_info(state: &crate::State) -> Vec<RpcPeerInfo> {
     state.p2p.ready().map_or_else(Vec::new, |p2p| {
         p2p.peers
             .iter()
             .map(|(peer_id, state)| {
                 let best_tip = state.status.as_ready().and_then(|r| r.best_tip.as_ref());
-                let (connection_status, time) = match &state.status {
+                let (connection_status, time, incoming) = match &state.status {
                     p2p::P2pPeerStatus::Connecting(c) => match c {
                         p2p::connection::P2pConnectionState::Outgoing(o) => {
-                            (PeerConnectionStatus::Connecting, o.time().into())
+                            (PeerConnectionStatus::Connecting, o.time().into(), false)
                         }
                         p2p::connection::P2pConnectionState::Incoming(i) => {
-                            (PeerConnectionStatus::Connecting, i.time().into())
+                            (PeerConnectionStatus::Connecting, i.time().into(), true)
                         }
                     },
                     p2p::P2pPeerStatus::Disconnecting { time } => {
-                        (PeerConnectionStatus::Disconnected, (*time).into())
+                        (PeerConnectionStatus::Disconnected, (*time).into(), false)
                     }
                     p2p::P2pPeerStatus::Disconnected { time } => {
-                        (PeerConnectionStatus::Disconnected, (*time).into())
-                    }
-                    p2p::P2pPeerStatus::Ready(r) => {
-                        (PeerConnectionStatus::Connected, r.connected_since.into())
+                        (PeerConnectionStatus::Disconnected, (*time).into(), false)
                     }
+                    p2p::P2pPeerStatus::Ready(r) => (
+                        PeerConnectionStatus::Connected,
+                        r.connected_since.into(),
+                        r.is_incoming,
+                    ),
                 };
                 RpcPeerInfo {
                     peer_id: *peer_id,
                     connection_status,
                     address: state.dial_opts.as_ref().map(|opts| opts.to_string()),
+                    incoming,
                     best_tip: best_tip.map(|bt| bt.hash.clone()),
                     best_tip_height: best_tip.map(|bt| bt.height()),
                     best_tip_global_slot: best_tip.map(|bt| bt.global_slot_since_genesis()),

node/src/rpc_effectful/rpc_effectful_effects.rs

@@ -1,15 +1,15 @@
-use super::RpcEffectfulAction;
+use super::{super::rpc, RpcEffectfulAction};
 use crate::{
     block_producer::BlockProducerWonSlot,
     external_snark_worker::available_job_to_snark_worker_spec,
     p2p::connection::P2pConnectionResponse,
     p2p_ready,
     rpc::{
         AccountQuery, AccountSlim, ActionStatsQuery, ActionStatsResponse, CurrentMessageProgress,
-        LedgerSyncProgress, MessagesStats, PeerConnectionStatus, RpcAction, RpcBlockProducerStats,
+        LedgerSyncProgress, MessagesStats, RpcAction, RpcBlockProducerStats,
         RpcMessageProgressResponse, RpcNodeStatus, RpcNodeStatusTransactionPool,
         RpcNodeStatusTransitionFrontier, RpcNodeStatusTransitionFrontierBlockSummary,
-        RpcNodeStatusTransitionFrontierSync, RpcPeerInfo, RpcRequestExtraData, RpcScanStateSummary,
+        RpcNodeStatusTransitionFrontierSync, RpcRequestExtraData, RpcScanStateSummary,
         RpcScanStateSummaryBlock, RpcScanStateSummaryBlockTransaction,
         RpcScanStateSummaryBlockTransactionKind, RpcScanStateSummaryScanStateJob,
         RpcSnarkPoolJobFull, RpcSnarkPoolJobSnarkWork, RpcSnarkPoolJobSummary,
@@ -72,7 +72,7 @@ pub fn rpc_effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta<RpcE
                         target: state.transition_frontier.sync.best_tip().map(block_summary),
                     },
                 },
-                peers: collect_rpc_peers_info(state),
+                peers: rpc::collect_rpc_peers_info(state),
                 snark_pool: state.snark_pool.jobs_iter().fold(
                     Default::default(),
                     |mut acc, job| {
@@ -709,43 +709,3 @@ pub fn rpc_effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta<RpcE
         }
     }
 }
-
-fn collect_rpc_peers_info(state: &crate::State) -> Vec<RpcPeerInfo> {
-    state.p2p.ready().map_or_else(Vec::new, |p2p| {
-        p2p.peers
-            .iter()
-            .map(|(peer_id, state)| {
-                let best_tip = state.status.as_ready().and_then(|r| r.best_tip.as_ref());
-                let (connection_status, time) = match &state.status {
-                    p2p::P2pPeerStatus::Connecting(c) => match c {
-                        p2p::connection::P2pConnectionState::Outgoing(o) => {
-                            (PeerConnectionStatus::Connecting, o.time().into())
-                        }
-                        p2p::connection::P2pConnectionState::Incoming(i) => {
-                            (PeerConnectionStatus::Connecting, i.time().into())
-                        }
-                    },
-                    p2p::P2pPeerStatus::Disconnecting { time } => {
-                        (PeerConnectionStatus::Disconnected, (*time).into())
-                    }
-                    p2p::P2pPeerStatus::Disconnected { time } => {
-                        (PeerConnectionStatus::Disconnected, (*time).into())
-                    }
-                    p2p::P2pPeerStatus::Ready(r) => {
-                        (PeerConnectionStatus::Connected, r.connected_since.into())
-                    }
-                };
-                RpcPeerInfo {
-                    peer_id: *peer_id,
-                    connection_status,
-                    address: state.dial_opts.as_ref().map(|opts| opts.to_string()),
-                    best_tip: best_tip.map(|bt| bt.hash.clone()),
-                    best_tip_height: best_tip.map(|bt| bt.height()),
-                    best_tip_global_slot: best_tip.map(|bt| bt.global_slot_since_genesis()),
-                    best_tip_timestamp: best_tip.map(|bt| bt.timestamp().into()),
-                    time,
-                }
-            })
-            .collect()
-    })
-}

p2p/src/connection/incoming/p2p_connection_incoming_reducer.rs

@@ -9,10 +9,11 @@ use crate::{
     connection::{
         incoming::P2pConnectionIncomingError,
         incoming_effectful::P2pConnectionIncomingEffectfulAction,
-        outgoing::P2pConnectionOutgoingInitOpts, P2pConnectionResponse, P2pConnectionState,
+        outgoing::{P2pConnectionOutgoingInitLibp2pOpts, P2pConnectionOutgoingInitOpts},
+        P2pConnectionResponse, P2pConnectionState,
     },
     disconnection::{P2pDisconnectionAction, P2pDisconnectionReason},
-    webrtc::{HttpSignalingInfo, SignalingMethod},
+    webrtc::{Host, HttpSignalingInfo, SignalingMethod},
     ConnectionAddr, P2pNetworkSchedulerAction, P2pPeerAction, P2pPeerState, P2pPeerStatus,
     P2pState, PeerId,
 };
@@ -416,7 +417,13 @@ impl P2pConnectionIncomingState {
                         .entry(peer_id)
                         .or_insert_with(|| P2pPeerState {
                             is_libp2p: true,
-                            dial_opts: None,
+                            dial_opts: Some(P2pConnectionOutgoingInitOpts::LibP2P(
+                                P2pConnectionOutgoingInitLibp2pOpts {
+                                    peer_id,
+                                    host: Host::from(addr.ip()),
+                                    port: addr.port(),
+                                },
+                            )),
                             status: P2pPeerStatus::Disconnected { time: meta.time() },
                             identify: None,
                         });