From dda743e1be05cfb80874cfb56426cb9c4b3aa2ac Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Sun, 2 Aug 2026 04:50:45 +0300 Subject: [PATCH 1/5] feat(node): Remove old P2P stack --- .../src/application/loops/connect_to_peers.rs | 699 ------------------ node/src/application/loops/main_loop.rs | 381 ---------- node/src/application/loops/mod.rs | 1 - node/src/application/loops/peer_loop.rs | 148 ++-- .../application/loops/peer_loop/channel.rs | 18 - node/src/application/network/bridge.rs | 2 +- node/src/application/network/codec.rs | 14 + node/src/application/network/mod.rs | 1 + node/src/lib.rs | 52 -- p2p/src/peer.rs | 11 - 10 files changed, 88 insertions(+), 1239 deletions(-) delete mode 100644 node/src/application/loops/connect_to_peers.rs create mode 100644 node/src/application/network/codec.rs diff --git a/node/src/application/loops/connect_to_peers.rs b/node/src/application/loops/connect_to_peers.rs deleted file mode 100644 index 16fce18..0000000 --- a/node/src/application/loops/connect_to_peers.rs +++ /dev/null @@ -1,699 +0,0 @@ -use std::collections::HashSet; -use std::fmt::Debug; -use std::net::IpAddr; -use std::net::SocketAddr; -use std::time::Duration; -use std::time::SystemTime; -use std::time::UNIX_EPOCH; - -use anyhow::bail; -use anyhow::ensure; -use anyhow::Result; -use bincode::Options; -use chrono::DateTime; -use chrono::Utc; -use futures::SinkExt; -use futures::TryStreamExt; -use libp2p::multiaddr::Protocol; -use libp2p::Multiaddr; -use tokio::io::AsyncRead; -use tokio::io::AsyncWrite; -use tokio::sync::broadcast; -use tokio::sync::mpsc; -use tokio::sync::OwnedSemaphorePermit; -use tokio::time::timeout; -use tokio_serde::formats::Bincode; -use tokio_serde::SymmetricallyFramed; -use tokio_util::codec::Framed; -use tokio_util::codec::LengthDelimitedCodec; -use tracing::debug; -use tracing::error; -use tracing::info; -use tracing::warn; - -use crate::application::config::cli_args; -use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; -use crate::application::config::parser::multiaddr::socketaddr_to_multiaddr; -use crate::application::loops::peer_loop::channel::MainToPeerTask; -use crate::application::loops::peer_loop::channel::PeerTaskToMain; -use crate::application::loops::peer_loop::PeerLoopHandler; -use crate::state::GlobalStateLock; -use crate::MAGIC_STRING_REQUEST; -use crate::MAGIC_STRING_RESPONSE; -use nyks_p2p::peer::handshake_data::HandshakeData; -use nyks_p2p::peer::handshake_data::VersionString; -use nyks_p2p::peer::peer_info::pseudorandom_peer_id; -use nyks_p2p::peer::ConnectionRefusedReason; -use nyks_p2p::peer::InternalConnectionStatus; -use nyks_p2p::peer::NegativePeerSanction; -use nyks_p2p::peer::PeerMessage; -use nyks_p2p::peer::PeerSanction; -use nyks_p2p::peer::PeerStanding; -use nyks_p2p::peer::TransferConnectionStatus; - -// Max peer message size is 500MB. Should be enough to send 250 blocks in a -// block batch-response. -pub const MAX_PEER_FRAME_LENGTH_IN_BYTES: usize = 500 * 1024 * 1024; - -/// Only accept connections where peer's reported timestamp deviates from our -/// by less than this value. -/// TODO: move to protocol... -const PEER_TIME_DIFFERENCE_THRESHOLD_IN_SECONDS: u128 = 90; - -/// Use this function to ensure that the same rules apply for both -/// ingoing and outgoing connections. This limits the size of messages -/// peers can send. -pub(crate) fn get_codec_rules() -> LengthDelimitedCodec { - let mut codec_rules = LengthDelimitedCodec::new(); - codec_rules.set_max_frame_length(MAX_PEER_FRAME_LENGTH_IN_BYTES); - codec_rules -} - -/// Returns a bincode codec with allocation limits to prevent OOM attacks. -/// -/// This prevents "length bomb" attacks where a malicious peer sends a tiny -/// frame claiming to contain a Vec with billions of elements. Without limits, -/// bincode would pre-allocate memory based on the claimed length, causing OOM. -/// -/// The limit is set to match MAX_PEER_FRAME_LENGTH_IN_BYTES (500MB). -fn get_bincode_codec() -> Bincode -where - Item: serde::de::DeserializeOwned, - SinkItem: serde::Serialize, -{ - bincode::DefaultOptions::new() - .with_limit(MAX_PEER_FRAME_LENGTH_IN_BYTES as u64) - .into() -} - -/// Infallible absolute difference between two timestamps, in seconds. -fn system_time_diff_seconds(peer: SystemTime, own: SystemTime) -> u128 { - let peer = peer - .duration_since(UNIX_EPOCH) - .map(|d| i128::from(d.as_secs())) - .unwrap_or_else(|e| -i128::from(e.duration().as_secs())); - - let own = own - .duration_since(UNIX_EPOCH) - .map(|d| i128::from(d.as_secs())) - .unwrap_or_else(|e| -i128::from(e.duration().as_secs())); - - (own - peer).unsigned_abs() -} - -/// Initial check if incoming connection is allowed. Performed prior to the -/// sending of the handshake. -pub(crate) fn precheck_incoming_connection_is_allowed( - cli: &cli_args::Args, - connecting_ip: IpAddr, -) -> bool { - let connecting_ip = match connecting_ip { - IpAddr::V4(_) => connecting_ip, - IpAddr::V6(v6) => match v6.to_ipv4() { - Some(v4) => std::net::IpAddr::V4(v4), - None => connecting_ip, - }, - }; - if cli.restrict_peers_to_list { - let allowed_ips: Vec = cli - .peers - .iter() - .filter_map(multiaddr_to_socketaddr) - .map(|p| p.ip()) - .collect(); - let is_allowed = allowed_ips.contains(&connecting_ip); - if !is_allowed { - debug!("Rejecting incoming connection from unlisted peer {connecting_ip} due to --restrict-peers-to-list",); - return false; - } - } - - if cli.banned_ips().contains(&connecting_ip) { - debug!("Rejecting incoming connection because it's explicitly banned"); - return false; - } - - if cli.disallow_all_incoming_peer_connections() { - debug!("Rejecting incoming connection because all incoming connections are disallowed"); - return false; - } - - true -} - -/// Check if connection is allowed. Used for both ingoing and outgoing connections. -/// -/// Note: this function is part of the legacy peer-to-peer stack. Therefore it -/// is okay to use [`SocketAddr`] and -/// [`pseudorandom_peer_id`]. -/// -/// # Locking -/// * acquires `global_state_lock` for read -async fn check_if_connection_is_allowed( - global_state_lock: GlobalStateLock, - own_handshake: &HandshakeData, - other_handshake: &HandshakeData, - peer_address: &SocketAddr, -) -> InternalConnectionStatus { - let cli_arguments = global_state_lock.cli(); - let global_state = global_state_lock.lock_guard().await; - - // Disallow connection if peer is banned via CLI arguments - if cli_arguments.banned_ips().contains(&peer_address.ip()) { - let ip = peer_address.ip(); - debug!("Peer {ip}, banned via CLI argument, attempted to connect. Disallowing."); - return InternalConnectionStatus::Refused(ConnectionRefusedReason::BadStanding); - } - - // Disallow connection if we disagree about time. - if system_time_diff_seconds(other_handshake.timestamp, own_handshake.timestamp) - > PEER_TIME_DIFFERENCE_THRESHOLD_IN_SECONDS - { - let own_datetime_utc: DateTime = own_handshake.timestamp.into(); - let peer_datetime_utc: DateTime = other_handshake.timestamp.into(); - warn!( - "New peer {} disagrees with us about time. Peer reports time {} but our clock at handshake was {}.", - peer_address, - peer_datetime_utc.format("%Y-%m-%d %H:%M:%S"), - own_datetime_utc.format("%Y-%m-%d %H:%M:%S")); - return InternalConnectionStatus::Refused(ConnectionRefusedReason::bad_timestamp()); - } - - // Disallow connection if peer is in bad standing - let standing = global_state - .net - .get_peer_standing_from_database(peer_address.ip()) - .await; - - // (But ignore bad standing if the peer is a CLI argument.) - let cli_peers = cli_arguments - .peers - .iter() - .filter_map(multiaddr_to_socketaddr) - .collect::>(); - if standing.is_some_and(|s| s.is_bad()) && !cli_peers.contains(peer_address) { - let ip = peer_address.ip(); - debug!("Peer {ip}, banned because of bad standing, attempted to connect. Disallowing."); - return InternalConnectionStatus::Refused(ConnectionRefusedReason::BadStanding); - } - - if let Some(time) = global_state - .net - .last_disconnection_time_of_peer(other_handshake.instance_id) - { - if SystemTime::now() - .duration_since(time) - .is_ok_and(|d| d < cli_arguments.reconnect_cooldown) - { - debug!( - "Refusing connection with {peer_address} \ - due to reconnect cooldown ({cooldown} seconds).", - cooldown = cli_arguments.reconnect_cooldown.as_secs(), - ); - - // A “wrong” reason is given because of backwards compatibility. - // todo: Use next breaking release to give a more accurate reason here. - let reason = ConnectionRefusedReason::MaxPeerNumberExceeded; - return InternalConnectionStatus::Refused(reason); - } - } - - // Disallow connection if max number of peers has been reached or - // exceeded. There is another test in `answer_peer_inner` that precedes - // this one; however this test is still necessary to resolve potential - // race conditions. - // Note that if we are bootstrapping, then we *do* want to accept the - // connection and temporarily exceed the maximum. In this case a - // `DisconnectFromLongestLivedPeer` message should have been sent to - // the main loop already but that message need not have been processed by - // the time we get here. - if cli_arguments.max_num_peers <= global_state.net.peer_map.len() && !cli_arguments.bootstrap { - return InternalConnectionStatus::Refused(ConnectionRefusedReason::MaxPeerNumberExceeded); - } - - // Disallow connection to already connected peer. - if global_state.net.peer_map.values().any(|peer| { - peer.instance_id() == other_handshake.instance_id - || multiaddr_to_socketaddr(&peer.address()).is_some_and(|sa| sa == *peer_address) - }) { - return InternalConnectionStatus::Refused(ConnectionRefusedReason::AlreadyConnected); - } - - // Cap connections per IP, if specified. - if let Some(max_connections_per_ip) = cli_arguments.max_connections_per_ip { - let peer_ip = peer_address.ip(); - let num_connections_to_this_ip = global_state - .net - .peer_map - .values() - .filter_map(|info| multiaddr_to_socketaddr(&info.address())) - .filter(|sa| sa.ip() == peer_ip) - .count(); - if num_connections_to_this_ip >= max_connections_per_ip { - return InternalConnectionStatus::Refused( - ConnectionRefusedReason::MaxPeerNumberExceeded, - ); - } - } - - // Disallow connection to self - if own_handshake.instance_id == other_handshake.instance_id { - return InternalConnectionStatus::Refused(ConnectionRefusedReason::SelfConnect); - } - - // Disallow connection if versions are incompatible - if !VersionString::versions_are_compatible(own_handshake.version, other_handshake.version) { - warn!( - "Attempting to connect to incompatible version. You might have to upgrade, or the other node does. Own version: {}, other version: {}", - own_handshake.version, - other_handshake.version); - return InternalConnectionStatus::Refused(ConnectionRefusedReason::IncompatibleVersion); - } - - // If this connection touches the maximum number of peer connections, say - // so with special OK code. - if cli_arguments.max_num_peers == global_state.net.peer_map.len() + 1 { - info!("ConnectionStatus::Accepted, but max # connections is now reached"); - return InternalConnectionStatus::AcceptedMaxReached; - } - - debug!("ConnectionStatus::Accepted"); - InternalConnectionStatus::Accepted -} - -/// Respond to an incoming connection initiation. -/// -/// Catch and process errors (if any) gracefully. -/// -/// All incoming connections from peers must go through this function. -/// -/// The `handshake_permit` is released after the handshake completes (success or -/// failure), not when the connection closes. This prevents semaphore starvation -/// attacks where an attacker holds connections idle to exhaust permits. -pub(crate) async fn answer_peer( - stream: S, - state_lock: GlobalStateLock, - peer_address: std::net::SocketAddr, - main_to_peer_task_rx: broadcast::Receiver, - peer_task_to_main_tx: mpsc::Sender, - own_handshake_data: HandshakeData, - handshake_permit: Option, -) -> Result<()> -where - S: AsyncRead + AsyncWrite + std::fmt::Debug + std::marker::Unpin, -{ - let inner_ret = answer_peer_inner( - stream, - state_lock.clone(), - peer_address, - main_to_peer_task_rx, - peer_task_to_main_tx, - own_handshake_data, - handshake_permit, - ) - .await; - - inner_ret -} - -/// Handles all* incoming connections. Returns when the connection is closed. -/// -/// A returned `Result::Error` always indicates that the connection was closed -/// through some networking error, either in this function or in the peer loop -/// that this function invokes if the handshake protocol is successful. -/// -/// A returned `Result::Ok` means that either the peer loop was entered or the -/// handshake was not completed. The reason for this behavior is that we want -/// malicious connection attempts to be as resource light as possible. And -/// allocating thousands of anyhow errors can use a lot of RAM. -/// -/// The `handshake_permit` is dropped after handshake completes to release the -/// semaphore slot for new incoming connection attempts. -/// -/// *: This function belongs to the legacy peer-to-peer stack, meaning in -/// particular that it is okay to use [`SocketAddr`]. However, this function -/// does not handle *all* incoming connections: connections coming in over the -/// libp2p network stack are handled by the -/// [`NetworkActor`](crate::application::network::actor::NetworkActor) and the -/// [`StreamGateway`](crate::application::network::gateway::StreamGateway). -async fn answer_peer_inner( - stream: S, - state: GlobalStateLock, - peer_address: SocketAddr, - main_to_peer_task_rx: broadcast::Receiver, - peer_task_to_main_tx: mpsc::Sender, - own_handshake_data: HandshakeData, - handshake_permit: Option, -) -> Result<()> -where - S: AsyncRead + AsyncWrite + Debug + Unpin, -{ - debug!("Established incoming TCP connection with {peer_address}"); - - // Build the communication/serialization/frame handler - let length_delimited = Framed::new(stream, get_codec_rules()); - let mut peer = SymmetricallyFramed::new(length_delimited, get_bincode_codec()); - - // Complete Neptune handshake - let handshake_timeout: u64 = state.cli().handshake_timeout.into(); - let handshake_timeout = Duration::from_secs(handshake_timeout); - let maybe_msg = timeout(handshake_timeout, peer.try_next()).await; - let (magic_value, peer_handshake) = match maybe_msg { - Ok(Ok(Some(PeerMessage::Handshake { magic_value, data }))) => (magic_value, data), - Ok(Ok(_)) => { - // no heavy anyhow::Error, just close - tracing::warn!(%peer_address, "unexpected message instead of handshake"); - return Ok(()); - } - Ok(Err(e)) => { - // no heavy anyhow::Error, just close - tracing::warn!(%peer_address, error = ?e, "I/O error during handshake"); - return Ok(()); - } - Err(_) => { - // no heavy anyhow::Error, just close - tracing::warn!(%peer_address, "handshake timed out"); - return Ok(()); - } - }; - - if magic_value != *MAGIC_STRING_REQUEST { - // no heavy anyhow::Error, just close - warn!("No valid magic value from {peer_address}. Closing connection."); - return Ok(()); - } - - let handshake_response = PeerMessage::Handshake { - magic_value: *MAGIC_STRING_RESPONSE, - data: Box::new(own_handshake_data), - }; - timeout(handshake_timeout, peer.send(handshake_response)).await??; - - // Verify peer network before moving on - let peer_network = peer_handshake.network; - let own_network = own_handshake_data.network; - ensure!( - peer_network == own_network, - "Cannot connect with {peer_address}: \ - Peer runs {peer_network}, this client runs {own_network}." - ); - - // Check if incoming connection is allowed - let connection_status = check_if_connection_is_allowed( - state.clone(), - &own_handshake_data, - &peer_handshake, - &peer_address, - ) - .await; - timeout( - handshake_timeout, - peer.send(PeerMessage::ConnectionStatus(connection_status.into())), - ) - .await??; - if let InternalConnectionStatus::Refused(reason) = connection_status { - let reason = format!("Refusing incoming connection. Reason: {reason:?}"); - debug!("{reason}"); - bail!("{reason}"); - } - - // Whether the incoming connection comes from a peer in bad standing is - // checked in `check_if_connection_is_allowed`. So if we get here, we are - // good to go. - info!("Connection accepted from {peer_address}"); - - // Release the handshake permit now that handshake is complete. - // This allows new incoming connections to start their handshake. - // The connection is now governed by max_num_peers, not the semaphore. - drop(handshake_permit); - - // If necessary, disconnect from another, existing peer. - if connection_status == InternalConnectionStatus::AcceptedMaxReached && state.cli().bootstrap { - info!("Maximum # peers reached, so disconnecting from an existing peer."); - peer_task_to_main_tx - .send(PeerTaskToMain::DisconnectFromLongestLivedPeer) - .await?; - } - - let peer_distance = 1; // All incoming connections have distance 1 - let peer_id = pseudorandom_peer_id(&peer_address); - let peer_multiaddr = socketaddr_to_multiaddr(peer_address); - let mut peer_loop_handler = PeerLoopHandler::new( - peer_task_to_main_tx, - state, - peer_id, - peer_multiaddr, - *peer_handshake, - true, - peer_distance, - ); - - // Run peer loop. - peer_loop_handler - .run_wrapper(peer, main_to_peer_task_rx) - .await?; - - Ok(()) -} - -/// Perform handshake and establish connection to a new peer while handling any -/// panics in the peer task gracefully. -/// -/// All* outgoing connections to peers must go through this function. -/// -/// *: This function belongs to the legacy peer-to-peer stack, which is in the -/// process of being deprecated. Since it is part of the legacy stack, it is -/// okay to use [`SocketAddr`]. However, the new libp2p network stack offers an -/// alternative way to call new peers, see -/// [`NetworkActorCommand::Dial`](crate::application::network::channel::NetworkActorCommand::Dial). -pub(crate) async fn call_peer( - peer_address: std::net::SocketAddr, - state: GlobalStateLock, - main_to_peer_task_rx: broadcast::Receiver, - peer_task_to_main_tx: mpsc::Sender, - own_handshake_data: HandshakeData, - peer_distance: u8, -) { - debug!("Attempting to initiate connection to {peer_address}"); - match tokio::net::TcpStream::connect(peer_address).await { - Err(e) => { - let msg = format!("Failed to establish TCP connection to {peer_address}: {e}"); - if peer_distance == 1 { - // outgoing connection to peer of distance 1 means user has - // requested a connection to this peer through CLI - // arguments, and should be warned if this fails. - warn!("{msg}"); - } else { - debug!("{msg}"); - } - } - Ok(stream) => { - match call_peer_inner( - stream, - state, - peer_address, - main_to_peer_task_rx, - peer_task_to_main_tx, - &own_handshake_data, - peer_distance, - ) - .await - { - Ok(()) => (), - Err(e) => { - let msg = format!("{e}. Failed to establish connection."); - // outgoing connection to peer of distance 1 means user has - // requested a connection to this peer through CLI - // arguments, and should be warned if this fails. - if peer_distance == 1 { - warn!("{msg}"); - } else { - debug!("{msg}"); - } - } - } - } - }; - - info!("Connection to {peer_address} closing"); -} - -/// Legacy peer-to-peer stack. -async fn call_peer_inner( - stream: S, - state: GlobalStateLock, - peer_address: std::net::SocketAddr, - main_to_peer_task_rx: broadcast::Receiver, - peer_task_to_main_tx: mpsc::Sender, - own_handshake: &HandshakeData, - peer_distance: u8, -) -> Result<()> -where - S: AsyncRead + AsyncWrite + Debug + Unpin, -{ - debug!("Established outgoing TCP connection with {peer_address}"); - - // Build the communication/serialization/frame handler - let length_delimited = Framed::new(stream, get_codec_rules()); - let mut peer = SymmetricallyFramed::new(length_delimited, get_bincode_codec()); - - // Make Neptune handshake - let outgoing_handshake = PeerMessage::Handshake { - magic_value: *MAGIC_STRING_REQUEST, - data: Box::new(own_handshake.to_owned()), - }; - peer.send(outgoing_handshake).await?; - debug!("Awaiting connection status response from {peer_address}"); - - let Some(PeerMessage::Handshake { - magic_value, - data: other_handshake, - }) = peer.try_next().await? - else { - bail!("Didn't get handshake response from {peer_address}"); - }; - ensure!( - magic_value == *MAGIC_STRING_RESPONSE, - "Didn't get expected magic value for handshake from {peer_address}", - ); - - debug!("Got correct magic value response from {peer_address}!"); - if other_handshake.network != own_handshake.network { - let other = other_handshake.network; - let own = own_handshake.network; - bail!("Cannot connect with {peer_address}: Peer runs {other}, this client runs {own}."); - } - - match peer.try_next().await? { - Some(PeerMessage::ConnectionStatus(TransferConnectionStatus::Accepted)) => { - debug!("Outgoing connection accepted by {peer_address}"); - } - Some(PeerMessage::ConnectionStatus(TransferConnectionStatus::Refused(reason))) => { - bail!("Outgoing connection attempt to {peer_address} refused. Reason: {reason:?}"); - } - _ => { - bail!( - "Got invalid connection status response from {peer_address} on outgoing connection" - ); - } - } - - // Peer accepted us. Check if we accept the peer. Note that the protocol does not stipulate - // that we answer with a connection status here, so if the connection is *not* accepted, we - // simply hang up but log the reason for the refusal. - let connection_status = check_if_connection_is_allowed( - state.clone(), - own_handshake, - &other_handshake, - &peer_address, - ) - .await; - if let InternalConnectionStatus::Refused(refused_reason) = connection_status { - warn!( - "Outgoing connection to {peer_address} refused. Reason: {:?}\nNow hanging up.", - refused_reason - ); - peer.send(PeerMessage::Bye).await?; - bail!("Attempted to connect to peer ({peer_address}) that was not allowed. This connection attempt should not have been made."); - } - - // By default, start by asking the peer for its peers. In an adversarial - // context, we want the network topology to be as robust as possible. - // Blockchain data can be obtained from other peers, if this connection - // fails. - peer.send(PeerMessage::PeerListRequest).await?; - - let peer_id = pseudorandom_peer_id(&peer_address); - let peer_multiaddr = socketaddr_to_multiaddr(peer_address); - let mut peer_loop_handler = PeerLoopHandler::new( - peer_task_to_main_tx, - state.clone(), - peer_id, - peer_multiaddr, - *other_handshake, - false, - peer_distance, - ); - - info!("Established outgoing connection to {peer_address}"); - - // Run peer loop. - peer_loop_handler - .run_wrapper(peer, main_to_peer_task_rx) - .await?; - - Ok(()) -} - -/// Remove peer from state. This function must be called every time -/// a peer is disconnected. Whether this happens through a panic -/// in the peer task or through a regular disconnect. -/// -/// This function is shared between the legacy peer-to-peer stack and the libp2p -/// network stack. -/// -/// Locking: -/// * acquires `global_state_lock` for write -pub(crate) async fn close_peer_connected_callback( - mut global_state_lock: GlobalStateLock, - peer_address: Multiaddr, - to_main_tx: &mpsc::Sender, -) { - let cli_arguments = global_state_lock.cli().clone(); - let mut global_state_mut = global_state_lock.lock_guard_mut().await; - - // Find the matching peer id - let Some(peer_id) = global_state_mut - .net - .peer_map - .iter() - .find(|(_peer_id, peer_info)| peer_info.address() == peer_address) - .map(|(peer_id, _)| peer_id) - .copied() - else { - error!("Could not find peer id for {peer_address}"); - return; - }; - - // Store any new peer-standing to database - let peer_info_writeback = global_state_mut.net.peer_map.remove(&peer_id); - let new_standing = if let Some(new) = peer_info_writeback { - new.standing() - } else { - error!("Could not find peer standing for {peer_address}"); - let mut standing = PeerStanding::new(cli_arguments.peer_tolerance); - let sanction = NegativePeerSanction::NoStandingFoundMaybeCrash; - - // Don't return early: _must_ send message to main loop at the end of this - // function. - // If the peer has now reached bad standing, the connection to it should be - // dropped, which is currently happening anyway. - let _ = standing.sanction(PeerSanction::Negative(sanction)); - standing - }; - debug!("Fetched peer info standing {new_standing} for peer {peer_address}"); - - let maybe_ip = peer_address.iter().find_map(|p| match p { - Protocol::Ip4(ip) => Some(IpAddr::V4(ip)), - Protocol::Ip6(ip) => Some(IpAddr::V6(ip)), - _ => None, - }); - if let Some(ip) = maybe_ip { - global_state_mut - .net - .write_peer_standing_on_decrease(ip, new_standing) - .await; - } - - let sync_mode_is_active = global_state_mut.net.sync_anchor.is_some(); - drop(global_state_mut); // avoid holding across mpsc::Sender::send() - debug!("Stored peer info standing {new_standing} for peer {peer_address}"); - - // If in sync mode, tell sync loop about dropped peer. - if sync_mode_is_active { - to_main_tx - .send(PeerTaskToMain::DroppedPeer(peer_id)) - .await - .expect("channel to main should exist"); - } -} diff --git a/node/src/application/loops/main_loop.rs b/node/src/application/loops/main_loop.rs index 3ae4ad0..259d327 100644 --- a/node/src/application/loops/main_loop.rs +++ b/node/src/application/loops/main_loop.rs @@ -35,9 +35,6 @@ use tracing::warn; use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; use crate::application::loops::channel::RPCServerToMain; -use crate::application::loops::connect_to_peers::answer_peer; -use crate::application::loops::connect_to_peers::call_peer; -use crate::application::loops::connect_to_peers::precheck_incoming_connection_is_allowed; use crate::application::loops::peer_loop::channel::MainToPeerTask; use crate::application::loops::peer_loop::channel::PeerTaskToMain; use crate::application::loops::sync_loop::channel::BlockRequest; @@ -64,7 +61,6 @@ pub(crate) const MAX_NUM_DIGESTS_IN_BATCH_REQUEST: usize = 200; /// MainLoop is the immutable part of the input for the main loop function #[derive(Debug)] pub struct MainLoopHandler { - incoming_peer_listener: TcpListener, global_state_lock: GlobalStateLock, // note: broadcast::Sender::send() does not block @@ -92,9 +88,6 @@ struct MutableMainLoopState { /// order. maybe_sync_loop: Option, - /// Information about potential peers for new connections. - potential_peers: PotentialPeersState, - /// A list of join-handles to spawned tasks. task_handles: Vec>, } @@ -103,154 +96,15 @@ impl MutableMainLoopState { fn new(task_handles: Vec>) -> Self { Self { maybe_sync_loop: None, - potential_peers: PotentialPeersState::default(), task_handles, } } } -/// holds information about a potential peer in the process of peer discovery -struct PotentialPeerInfo { - _reported: SystemTime, - _reported_by: PeerId, - instance_id: u128, - distance: u8, -} - -impl PotentialPeerInfo { - fn new(reported_by: PeerId, instance_id: u128, distance: u8, now: SystemTime) -> Self { - Self { - _reported: now, - _reported_by: reported_by, - instance_id, - distance, - } - } -} - -/// holds information about a set of potential peers in the process of peer discovery -struct PotentialPeersState { - potential_peers: HashMap, -} - -impl PotentialPeersState { - fn default() -> Self { - Self { - potential_peers: HashMap::new(), - } - } - - fn add( - &mut self, - reported_by: PeerId, - potential_peer: (SocketAddr, u128), - max_peers: usize, - distance: u8, - now: SystemTime, - ) { - let potential_peer_socket_address = potential_peer.0; - let potential_peer_instance_id = potential_peer.1; - - // This check *should* make it likely that a potential peer is always - // registered with the lowest observed distance. - if self - .potential_peers - .contains_key(&potential_peer_socket_address) - { - return; - } - - // If this data structure is full, remove a random entry. Then add this. - if self.potential_peers.len() - > max_peers * POTENTIAL_PEER_MAX_COUNT_AS_A_FACTOR_OF_MAX_PEERS - { - let mut rng = rand::rng(); - let random_potential_peer = self - .potential_peers - .keys() - .choose(&mut rng) - .unwrap() - .to_owned(); - self.potential_peers.remove(&random_potential_peer); - } - - let insert_value = - PotentialPeerInfo::new(reported_by, potential_peer_instance_id, distance, now); - self.potential_peers - .insert(potential_peer_socket_address, insert_value); - } - - /// Return a peer from the potential peer list that we aren't connected to - /// and that isn't our own address. - /// - /// Favors peers with a high distance and with IPs that we are not already - /// connected to. - /// - /// Returns (socket address, peer distance) - /// - /// This function is part of the legacy peer-to-peer stack. - fn get_candidate( - &self, - connected_clients: &[PeerInfo], - own_instance_id: u128, - ) -> Option<(SocketAddr, u8)> { - let peers_instance_ids: Vec = - connected_clients.iter().map(|x| x.instance_id()).collect(); - - // Only pick those peers that report a listening port - let peers_listen_addresses: Vec = connected_clients - .iter() - .filter_map(|x| x.listen_address()) - .filter_map(|ma| multiaddr_to_socketaddr(&ma)) - .collect(); - - // Find the appropriate candidates - let candidates = self - .potential_peers - .iter() - // Prevent connecting to self. Note that we *only* use instance ID to prevent this, - // meaning this will allow multiple nodes e.g. running on the same computer to form - // a complete graph. - .filter(|pp| pp.1.instance_id != own_instance_id) - // Prevent connecting to peer we already are connected to - .filter(|potential_peer| !peers_instance_ids.contains(&potential_peer.1.instance_id)) - .filter(|potential_peer| !peers_listen_addresses.contains(potential_peer.0)) - .collect::>(); - - // Prefer candidates with IPs that we are not already connected to but - // connect to repeated IPs in case we don't have other options, as - // repeated IPs may just be multiple machines on the same NAT'ed IPv4 - // address. - let mut connected_ips = peers_listen_addresses.into_iter().map(|x| x.ip()); - let candidates = if candidates - .iter() - .any(|candidate| !connected_ips.contains(&candidate.0.ip())) - { - candidates - .into_iter() - .filter(|candidate| !connected_ips.contains(&candidate.0.ip())) - .collect() - } else { - candidates - }; - - // Get the candidate list with the highest distance - let max_distance_candidates = candidates.iter().max_by_key(|pp| pp.1.distance); - - // Pick a random candidate from the appropriate candidates - let mut rng = rand::rng(); - max_distance_candidates - .iter() - .choose(&mut rng) - .map(|x| (x.0.to_owned(), x.1.distance)) - } -} - impl MainLoopHandler { // todo: find a way to avoid triggering lint #[expect(clippy::too_many_arguments)] pub(crate) fn new( - incoming_peer_listener: TcpListener, global_state_lock: GlobalStateLock, main_to_peer_broadcast_tx: broadcast::Sender, peer_task_to_main_tx: mpsc::Sender, @@ -262,7 +116,6 @@ impl MainLoopHandler { task_handles: Vec>, ) -> Self { Self { - incoming_peer_listener, global_state_lock, main_to_peer_broadcast_tx, peer_task_to_main_tx, @@ -514,20 +367,6 @@ impl MainLoopHandler { } } } - PeerTaskToMain::PeerDiscoveryAnswer((pot_peers, reported_by, distance)) => { - log_slow_scope!(fn_name!() + "::PeerTaskToMain::PeerDiscoveryAnswer"); - - let max_peers = self.global_state_lock.cli().max_num_peers; - for pot_peer in pot_peers { - main_loop_state.potential_peers.add( - reported_by, - pot_peer, - max_peers, - distance, - self.now(), - ); - } - } PeerTaskToMain::NewPeer(peer_id) => { if let Some(sync_loop) = &main_loop_state.maybe_sync_loop { sync_loop.send_add_peer(peer_id).await; @@ -774,162 +613,6 @@ impl MainLoopHandler { Ok(()) } - /// If necessary, reconnect to the peers listed as CLI arguments. - /// - /// Locking: - /// * acquires `global_state_lock` for read - async fn reconnect(&self, main_loop_state: &mut MutableMainLoopState) -> Result<()> { - let connected_peers = self - .global_state_lock - .lock_guard() - .await - .net - .peer_map - .iter() - .map(|(peer_id, peer_info)| (*peer_id, peer_info.clone())) - .collect_vec(); - let connected_peers_addresses = connected_peers - .iter() - .map(|(_peer_id, peer_info)| peer_info.address().clone()) - .collect_vec(); - let peers_with_lost_connection = self - .global_state_lock - .cli() - .peers - .iter() - .filter(|peer| !connected_peers_addresses.contains(peer)); - - // If no connection was lost, there's nothing to do. - if peers_with_lost_connection.clone().count() == 0 { - return Ok(()); - } - - // Else, try to reconnect. - let own_handshake_data = self - .global_state_lock - .lock_guard() - .await - .get_own_handshakedata(); - for peer_with_lost_connection in peers_with_lost_connection { - if let Some(socketaddr) = multiaddr_to_socketaddr(peer_with_lost_connection) { - // Disallow reconnection if peer is in bad standing - let peer_standing = self - .global_state_lock - .lock_guard() - .await - .net - .get_peer_standing_from_database(socketaddr.ip()) - .await; - if peer_standing.is_some_and(|standing| standing.is_bad()) { - debug!("Not reconnecting to peer in bad standing: {socketaddr}"); - continue; - } - - debug!("Attempting to reconnect to peer: {socketaddr}"); - let global_state_lock = self.global_state_lock.clone(); - let main_to_peer_broadcast_rx = self.main_to_peer_broadcast_tx.subscribe(); - let peer_task_to_main_tx = self.peer_task_to_main_tx.to_owned(); - let outgoing_connection_task = tokio::task::spawn(async move { - call_peer( - socketaddr, - global_state_lock, - main_to_peer_broadcast_rx, - peer_task_to_main_tx, - own_handshake_data, - 1, // All CLI-specified peers have distance 1 - ) - .await; - }); - main_loop_state.task_handles.push(outgoing_connection_task); - } else if let Err(e) = self - .network_command_tx - .send(NetworkActorCommand::Dial(peer_with_lost_connection.clone())) - .await - { - warn!("Failed to reconnect to peer {peer_with_lost_connection}: {e}."); - } - main_loop_state.task_handles.retain(|th| !th.is_finished()); - } - - Ok(()) - } - - /// Perform peer discovery. - /// - /// Peer discovery involves finding potential peers from connected peers - /// and attempts to establish a connection with one of them. - /// - /// Locking: - /// * acquires `global_state_lock` for read - async fn discover_peers(&self, main_loop_state: &mut MutableMainLoopState) -> Result<()> { - // fetch all relevant info from global state, then release the lock - let cli_args = self.global_state_lock.cli(); - let global_state = self.global_state_lock.lock_guard().await; - let connected_peers = global_state.net.peer_map.values().cloned().collect_vec(); - let own_instance_id = global_state.net.instance_id; - let own_handshake_data = global_state.get_own_handshakedata(); - drop(global_state); - - let num_peers = connected_peers.len(); - let max_num_peers = cli_args.max_num_peers; - - // Don't make an outgoing connection if - // - the peer limit is reached (or exceeded), or - // - the peer limit is _almost_ reached; reserve the last slot for an - // incoming connection. - if num_peers >= max_num_peers || num_peers > 2 && num_peers - 1 == max_num_peers { - debug!("Connected to {num_peers} peers. The configured max is {max_num_peers} peers."); - debug!("Skipping peer discovery."); - return Ok(()); - } - - debug!("Performing peer discovery"); - - // Ask all peers for their peer lists. This will eventually – once the - // responses have come in – update the list of potential peers. - let pmsg = MainToPeerTask::MakePeerDiscoveryRequest; - self.main_to_peer_broadcast(pmsg); - - // Get a peer candidate from the list of potential peers. Generally, - // the peer lists requested in the previous step will not have come in - // yet. Therefore, the new candidate is selected based on somewhat - // (but not overly) old information. - let Some((peer_candidate, candidate_distance)) = main_loop_state - .potential_peers - .get_candidate(&connected_peers, own_instance_id) - else { - debug!("Found no peer candidate to connect to. Not making new connection."); - return Ok(()); - }; - - // Try to connect to the selected candidate. - debug!("Connecting to peer {peer_candidate} with distance {candidate_distance}"); - let global_state_lock = self.global_state_lock.clone(); - let main_to_peer_broadcast_rx = self.main_to_peer_broadcast_tx.subscribe(); - let peer_task_to_main_tx = self.peer_task_to_main_tx.to_owned(); - let outgoing_connection_task = tokio::task::spawn(async move { - call_peer( - peer_candidate, - global_state_lock, - main_to_peer_broadcast_rx, - peer_task_to_main_tx, - own_handshake_data, - candidate_distance, - ) - .await; - }); - main_loop_state.task_handles.push(outgoing_connection_task); - main_loop_state.task_handles.retain(|th| !th.is_finished()); - - // Immediately request the new peer's peer list. This allows - // incorporating the new peer's peers into the list of potential peers, - // to be used in the next round of peer discovery. - let m2pmsg = MainToPeerTask::MakeSpecificPeerDiscoveryRequest(peer_candidate); - self.main_to_peer_broadcast(m2pmsg); - - Ok(()) - } - pub async fn run(&mut self) -> Result { info!("Starting main loop"); @@ -1002,16 +685,6 @@ impl MainLoopHandler { #[cfg(not(unix))] drop((tx_term, tx_int, tx_quit)); - // Use a semaphore to limit number of incoming connections. Should only be - // relevant as a countermeasure against a DOS. Each incoming connection - // must acquire a permit. If none is free, the below call to `acquire_owned` - // will only be resolved when an incoming connection is closed. This - // value is set much higher than the configured max number of peers since it's - // only intended to be used in case of heavy DOS. - let incoming_connections_limit = Arc::new(Semaphore::new( - self.global_state_lock.cli().max_num_peers * 2 + 4, - )); - let exit_code: i32 = loop { select! { Ok(()) = signal::ctrl_c() => { @@ -1033,58 +706,6 @@ impl MainLoopHandler { break SUCCESS_EXIT_CODE; } - // Handle incoming connections from peer - Ok((stream, peer_address)) = self.incoming_peer_listener.accept() => { - let ip = peer_address.ip(); - if !precheck_incoming_connection_is_allowed(self.global_state_lock.cli(), ip) { - continue; - } - - // Is this IP banned through database entry? - let peer_banned = self.global_state_lock.lock_guard().await.net.peer_databases.peer_standings_by_ip.get(ip).await.is_some_and(|x| x.is_bad()); - if peer_banned { - debug!("Banned peer {ip} attempted incoming connection. Hanging up."); - continue; - } - - // Bump semaphore counter for incoming connections. Should - // be done after the precheck to prevent unnecessary - // acquisitions. - let timeout = Duration::from_secs(self.global_state_lock.cli().handshake_timeout.into()); - let permit = time::timeout(timeout, incoming_connections_limit.clone().acquire_owned()).await; - let Ok(permit) = permit else { - warn!("Too many incoming connections to handle. Dropping incoming connection from {ip}."); - continue; - }; - - let permit = permit?; - - let state = self.global_state_lock.lock_guard().await; - let main_to_peer_broadcast_rx_clone: broadcast::Receiver = self.main_to_peer_broadcast_tx.subscribe(); - let peer_task_to_main_tx_clone: mpsc::Sender = self.peer_task_to_main_tx.clone(); - let own_handshake_data: HandshakeData = state.get_own_handshakedata(); - let global_state_lock = self.global_state_lock.clone(); // bump arc refcount. - let incoming_peer_task_handle = tokio::task::spawn(async move { - // Permit is passed to answer_peer and released after handshake, - // not when the connection closes. This prevents semaphore - // starvation attacks. - match answer_peer( - stream, - global_state_lock, - peer_address, - main_to_peer_broadcast_rx_clone, - peer_task_to_main_tx_clone, - own_handshake_data, - Some(permit), - ).await { - Ok(()) => (), - Err(err) => debug!("Got result: {:?}", err), - } - }); - main_loop_state.task_handles.push(incoming_peer_task_handle); - main_loop_state.task_handles.retain(|th| !th.is_finished()); - } - // Handle messages from peer tasks Some(msg) = self.peer_task_to_main_rx.recv() => { debug!("Received message sent to main task."); @@ -1142,8 +763,6 @@ impl MainLoopHandler { if perform_discovery { self.prune_peers().await?; - self.reconnect(&mut main_loop_state).await?; - self.discover_peers(&mut main_loop_state).await?; } } diff --git a/node/src/application/loops/mod.rs b/node/src/application/loops/mod.rs index 4689a76..ee77627 100644 --- a/node/src/application/loops/mod.rs +++ b/node/src/application/loops/mod.rs @@ -1,5 +1,4 @@ pub mod channel; -pub mod connect_to_peers; pub mod main_loop; pub mod peer_loop; pub mod sync_loop; diff --git a/node/src/application/loops/peer_loop.rs b/node/src/application/loops/peer_loop.rs index 9022f5f..42442c2 100644 --- a/node/src/application/loops/peer_loop.rs +++ b/node/src/application/loops/peer_loop.rs @@ -3,7 +3,6 @@ pub(crate) mod channel; use std::cmp; use std::marker::Unpin; use std::net::IpAddr; -use std::net::SocketAddr; use std::time::SystemTime; use anyhow::bail; @@ -56,8 +55,6 @@ use tracing::error; use tracing::info; use tracing::warn; -use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; -use crate::application::loops::connect_to_peers::close_peer_connected_callback; use crate::application::loops::main_loop::MAX_NUM_DIGESTS_IN_BATCH_REQUEST; use crate::application::loops::peer_loop::channel::MainToPeerTask; use crate::application::loops::peer_loop::channel::PeerTaskToMain; @@ -71,7 +68,6 @@ use crate::state::GlobalState; use crate::state::GlobalStateLock; const STANDARD_BLOCK_BATCH_SIZE: usize = 35; -const MAX_PEER_LIST_LENGTH: usize = 10; const MINIMUM_BLOCK_BATCH_SIZE: usize = 2; /// Maximum size in bytes for a single block during fork reconciliation. Blocks @@ -618,66 +614,6 @@ impl PeerLoopHandler { info!("Got bye. Closing connection to peer"); Ok(DISCONNECT_CONNECTION) } - PeerMessage::PeerListRequest => { - let peer_info = { - log_slow_scope!(fn_name!() + "::PeerMessage::PeerListRequest"); - - // We are interested in the address on which peers accept ingoing connections, - // not in the address in which they are connected to us. We are only interested in - // peers that accept incoming connections. - let mut peer_info: Vec<(SocketAddr, u128)> = self - .global_state_lock - .lock_guard() - .await - .net - .peer_map - .values() - .filter(|peer_info| { - peer_info.listen_address().is_some() && !peer_info.is_local_connection() - }) - .take(MAX_PEER_LIST_LENGTH) // limit length of response - .filter_map(|peer_info| { - multiaddr_to_socketaddr( - &peer_info - .listen_address() - .expect("already filtered for some listen address"), - ) - .map(|socket_addr| (socket_addr, peer_info.instance_id())) - }) - .collect(); - - // We sort the returned list, so this function is easier to test - peer_info.sort_by_cached_key(|x| x.0); - peer_info - }; - - debug!("Responding with: {:?}", peer_info); - peer.send(PeerMessage::PeerListResponse(peer_info)).await?; - Ok(KEEP_CONNECTION_ALIVE) - } - PeerMessage::PeerListResponse(peers) => { - log_slow_scope!(fn_name!() + "::PeerMessage::PeerListResponse"); - - if peers.len() > MAX_PEER_LIST_LENGTH { - self.punish(NegativePeerSanction::FloodPeerListResponse) - .await?; - } - - let peers = peers - .into_iter() - .filter(|(socket_addr, _)| !PeerInfo::ip_is_local(socket_addr.ip())) - .collect(); - - self.to_main_tx - .send(PeerTaskToMain::PeerDiscoveryAnswer(( - peers, - self.peer_id, - // The distance to the revealed peers is 1 + this peer's distance - self.distance + 1, - ))) - .await?; - Ok(KEEP_CONNECTION_ALIVE) - } PeerMessage::BlockNotificationRequest => { debug!("Got BlockNotificationRequest"); @@ -1972,10 +1908,6 @@ impl PeerLoopHandler { // sanction, we don't disconnect. Ok(KEEP_CONNECTION_ALIVE) } - MainToPeerTask::MakePeerDiscoveryRequest => { - peer.send(PeerMessage::PeerListRequest).await?; - Ok(KEEP_CONNECTION_ALIVE) - } MainToPeerTask::Disconnect(peer_id) => { log_slow_scope!(fn_name!() + "::MainToPeerTask::Disconnect"); @@ -1997,14 +1929,6 @@ impl PeerLoopHandler { Ok(DISCONNECT_CONNECTION) } - MainToPeerTask::MakeSpecificPeerDiscoveryRequest(target_socket_addr) => { - if let Some(socket_addr) = multiaddr_to_socketaddr(&self.peer_address) { - if target_socket_addr == socket_addr { - peer.send(PeerMessage::PeerListRequest).await?; - } - } - Ok(KEEP_CONNECTION_ALIVE) - } MainToPeerTask::TransactionNotification(transaction_notification) => { debug!("Sending PeerMessage::TransactionNotification"); peer.send(PeerMessage::TransactionNotification( @@ -2326,3 +2250,75 @@ impl PeerLoopHandler { .register_peer_disconnection(peer_id, SystemTime::now()); } } + +/// Remove peer from state. This function must be called every time +/// a peer is disconnected. Whether this happens through a panic +/// in the peer task or through a regular disconnect. +/// +/// TODO: make this part of handler? +/// +/// Locking: +/// * acquires `global_state_lock` for write +pub(crate) async fn close_peer_connected_callback( + mut global_state_lock: GlobalStateLock, + peer_address: Multiaddr, + to_main_tx: &mpsc::Sender, +) { + let cli_arguments = global_state_lock.cli().clone(); + let mut global_state_mut = global_state_lock.lock_guard_mut().await; + + // Find the matching peer id + let Some(peer_id) = global_state_mut + .net + .peer_map + .iter() + .find(|(_peer_id, peer_info)| peer_info.address() == peer_address) + .map(|(peer_id, _)| peer_id) + .copied() + else { + error!("Could not find peer id for {peer_address}"); + return; + }; + + // Store any new peer-standing to database + let peer_info_writeback = global_state_mut.net.peer_map.remove(&peer_id); + let new_standing = if let Some(new) = peer_info_writeback { + new.standing() + } else { + error!("Could not find peer standing for {peer_address}"); + let mut standing = PeerStanding::new(cli_arguments.peer_tolerance); + let sanction = NegativePeerSanction::NoStandingFoundMaybeCrash; + + // Don't return early: _must_ send message to main loop at the end of this + // function. + // If the peer has now reached bad standing, the connection to it should be + // dropped, which is currently happening anyway. + let _ = standing.sanction(PeerSanction::Negative(sanction)); + standing + }; + debug!("Fetched peer info standing {new_standing} for peer {peer_address}"); + + let maybe_ip = peer_address.iter().find_map(|p| match p { + Protocol::Ip4(ip) => Some(IpAddr::V4(ip)), + Protocol::Ip6(ip) => Some(IpAddr::V6(ip)), + _ => None, + }); + if let Some(ip) = maybe_ip { + global_state_mut + .net + .write_peer_standing_on_decrease(ip, new_standing) + .await; + } + + let sync_mode_is_active = global_state_mut.net.sync_anchor.is_some(); + drop(global_state_mut); // avoid holding across mpsc::Sender::send() + debug!("Stored peer info standing {new_standing} for peer {peer_address}"); + + // If in sync mode, tell sync loop about dropped peer. + if sync_mode_is_active { + to_main_tx + .send(PeerTaskToMain::DroppedPeer(peer_id)) + .await + .expect("channel to main should exist"); + } +} diff --git a/node/src/application/loops/peer_loop/channel.rs b/node/src/application/loops/peer_loop/channel.rs index 7efad7c..876d44b 100644 --- a/node/src/application/loops/peer_loop/channel.rs +++ b/node/src/application/loops/peer_loop/channel.rs @@ -1,5 +1,3 @@ -use std::net::SocketAddr; - use libp2p::Multiaddr; use libp2p::PeerId; use nyks_p2p::peer::peer_block_notifications::BlockProposalNotification; @@ -26,12 +24,6 @@ pub(crate) enum MainToPeerTask { /// sanction a peer for failing to respond to sync request PeerSynchronizationTimeout(PeerId), - /// Request peer list from connected peers - MakePeerDiscoveryRequest, - - /// Request peers from a specific peer to get peers further away - MakeSpecificPeerDiscoveryRequest(SocketAddr), - /// Publish knowledge of a transaction TransactionNotification(TransactionNotification), @@ -61,10 +53,6 @@ impl MainToPeerTask { MainToPeerTask::Block(_) => "block", MainToPeerTask::RequestBlockByHeight { .. } => "req block by height", MainToPeerTask::PeerSynchronizationTimeout(_) => "peer sync timeout", - MainToPeerTask::MakePeerDiscoveryRequest => "make peer discovery req", - MainToPeerTask::MakeSpecificPeerDiscoveryRequest(_) => { - "make specific peer discovery req" - } MainToPeerTask::TransactionNotification(_) => "transaction notification", MainToPeerTask::Disconnect(_) => "disconnect", MainToPeerTask::DisconnectAll() => "disconnect all", @@ -84,8 +72,6 @@ impl MainToPeerTask { MainToPeerTask::BlockProposalNotification(_) => true, MainToPeerTask::RequestBlockByHeight { .. } => true, MainToPeerTask::PeerSynchronizationTimeout(_) => true, - MainToPeerTask::MakePeerDiscoveryRequest => false, - MainToPeerTask::MakeSpecificPeerDiscoveryRequest(_) => false, MainToPeerTask::TransactionNotification(_) => true, MainToPeerTask::Disconnect(_) => false, MainToPeerTask::DisconnectAll() => false, @@ -111,9 +97,6 @@ pub(crate) enum PeerTaskToMain { claimed_block_digest: Digest, }, - /// (\[(peer_listen_address)\], reported_by, distance) - PeerDiscoveryAnswer((Vec<(SocketAddr, u128)>, PeerId, u8)), - Transaction(Box), BlockProposal(Box), DisconnectFromLongestLivedPeer, @@ -143,7 +126,6 @@ impl PeerTaskToMain { match self { PeerTaskToMain::NewBlocks(_) => "new blocks", PeerTaskToMain::AddPeerMaxBlockHeight { .. } => "add peer max block height", - PeerTaskToMain::PeerDiscoveryAnswer(_) => "peer discovery answer", PeerTaskToMain::Transaction(_) => "transaction", PeerTaskToMain::BlockProposal(_) => "block proposal", PeerTaskToMain::DisconnectFromLongestLivedPeer => "disconnect from longest lived peer", diff --git a/node/src/application/network/bridge.rs b/node/src/application/network/bridge.rs index 5cae2d8..a5024af 100644 --- a/node/src/application/network/bridge.rs +++ b/node/src/application/network/bridge.rs @@ -10,7 +10,7 @@ use nyks_p2p::peer::PeerMessage; use tokio_util::codec::Framed; use tokio_util::compat::FuturesAsyncReadCompatExt; -use crate::application::loops::connect_to_peers::get_codec_rules; +use crate::application::network::codec::get_codec_rules; /// A transport-agnostic wrapper for peer communication. /// diff --git a/node/src/application/network/codec.rs b/node/src/application/network/codec.rs new file mode 100644 index 0000000..d38401a --- /dev/null +++ b/node/src/application/network/codec.rs @@ -0,0 +1,14 @@ +use tokio_util::codec::LengthDelimitedCodec; + +// Max peer message size is 500MB. Should be enough to send 250 blocks in a +// block batch-response. +pub const MAX_PEER_FRAME_LENGTH_IN_BYTES: usize = 500 * 1024 * 1024; + +/// Use this function to ensure that the same rules apply for both +/// ingoing and outgoing connections. This limits the size of messages +/// peers can send. +pub(crate) fn get_codec_rules() -> LengthDelimitedCodec { + let mut codec_rules = LengthDelimitedCodec::new(); + codec_rules.set_max_frame_length(MAX_PEER_FRAME_LENGTH_IN_BYTES); + codec_rules +} \ No newline at end of file diff --git a/node/src/application/network/mod.rs b/node/src/application/network/mod.rs index 958d805..46c9964 100644 --- a/node/src/application/network/mod.rs +++ b/node/src/application/network/mod.rs @@ -9,6 +9,7 @@ pub(crate) mod handshake; pub mod overview; pub(crate) mod reachability; pub(crate) mod stack; +pub(crate) mod codec; #[cfg(any(test, feature = "arbitrary-impls"))] pub mod arbitrary; diff --git a/node/src/lib.rs b/node/src/lib.rs index fdaf9ff..0745771 100644 --- a/node/src/lib.rs +++ b/node/src/lib.rs @@ -54,7 +54,6 @@ use crate::application::config::data_directory::DataDirectory; use crate::application::config::identity::resolve_identity; use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; use crate::application::loops::channel::RPCServerToMain; -use crate::application::loops::connect_to_peers::call_peer; use crate::application::loops::main_loop::MainLoopHandler; use crate::application::loops::peer_loop::channel::MainToPeerTask; use crate::application::loops::peer_loop::channel::PeerTaskToMain; @@ -69,10 +68,6 @@ use crate::state::GlobalStateLock; pub const SUCCESS_EXIT_CODE: i32 = 0; pub const COMPOSITION_FAILED_EXIT_CODE: i32 = 159; -/// Magic string to ensure other program is Neptune Core -/// TODO: move to protocol -pub const MAGIC_STRING_REQUEST: &[u8; 15] = b"7B8AB7FC438F411"; -pub const MAGIC_STRING_RESPONSE: &[u8; 15] = b"Hello Neptune!\n"; const PEER_CHANNEL_CAPACITY: usize = 1000; const RPC_CHANNEL_CAPACITY: usize = 1000; const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -138,11 +133,9 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { // Set up the libp2p NetworkActor info!("Setting up Network Actor"); - let legacy_marker = libp2p::multiaddr::Protocol::Tcp(9798); let cli_peers_for_network_actor = cli_args .peers .iter() - .filter(|addr| addr.iter().all(|p| p != legacy_marker)) .cloned() .collect_vec(); let network_config = NetworkConfig::default() @@ -198,18 +191,6 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { } } - // Bind socket to port on this machine, to handle incoming connections from peers - let incoming_peer_listener = if let Some(incoming_peer_listener) = cli_args.own_listen_port() { - let ret = TcpListener::bind((cli_args.peer_listen_addr, incoming_peer_listener)) - .await - .with_context(|| format!("Failed to bind to local TCP port {}:{}. Is an instance of this program already running?", cli_args.peer_listen_addr, incoming_peer_listener))?; - info!("Now listening for incoming peer-connections"); - ret - } else { - info!("Not accepting incoming peer-connections"); - TcpListener::bind("127.0.0.1:0").await? - }; - // Connect to peers, and provide each peer task with a thread-safe copy of the state let own_handshake_data: HandshakeData = global_state_lock.lock_guard().await.get_own_handshakedata(); @@ -217,38 +198,6 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { "Most known canonical block has height {}", own_handshake_data.tip_header.height ); - let legacy_socketaddr = |multiaddr: &libp2p::Multiaddr| { - multiaddr_to_socketaddr(multiaddr).and_then(|sa| { - if [9800, 9801].contains(&sa.port()) { - None - } else { - Some(sa) - } - }) - }; - for multiaddress in &global_state_lock.cli().peers { - if let Some(peer_address) = legacy_socketaddr(multiaddress) { - let peer_state_var = global_state_lock.clone(); // bump arc refcount - let main_to_peer_broadcast_rx_clone: broadcast::Receiver = - main_to_peer_broadcast_tx.subscribe(); - let peer_task_to_main_tx_clone: mpsc::Sender = - peer_task_to_main_tx.clone(); - let peer_join_handle = tokio::task::spawn(async move { - call_peer( - peer_address, - peer_state_var.clone(), - main_to_peer_broadcast_rx_clone, - peer_task_to_main_tx_clone, - own_handshake_data, - 1, // All outgoing connections have distance 1 - ) - .await; - }); - task_join_handles.push(peer_join_handle); - } - // Else: NetworkActor already got CLI peers via NetworkConfig. - } - debug!("Made outgoing connections to peers"); // Start RPC server for CLI request and more. It's important that this is done as late // as possible, so requests do not hang while initialization code runs. @@ -267,7 +216,6 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { // Handle incoming connections, messages from peer tasks, and messages from the mining task Ok(MainLoopHandler::new( - incoming_peer_listener, global_state_lock, main_to_peer_broadcast_tx, peer_task_to_main_tx, diff --git a/p2p/src/peer.rs b/p2p/src/peer.rs index fd1b905..32db628 100644 --- a/p2p/src/peer.rs +++ b/p2p/src/peer.rs @@ -478,9 +478,6 @@ pub enum PeerMessage { /// Send a request that this node would like a copy of the transaction with /// digest as specified by the argument. TransactionRequest(TransactionKernelId), - PeerListRequest, - /// (socket address, instance_id) - PeerListResponse(Vec<(SocketAddr, u128)>), /// Inform peer that we are disconnecting them. Bye, ConnectionStatus(TransferConnectionStatus), @@ -502,8 +499,6 @@ impl PeerMessage { PeerMessage::Transaction(_) => "send", PeerMessage::TransactionNotification(_) => "transaction notification", PeerMessage::TransactionRequest(_) => "transaction request", - PeerMessage::PeerListRequest => "peer list req", - PeerMessage::PeerListResponse(_) => "peer list resp", PeerMessage::Bye => "bye", PeerMessage::ConnectionStatus(_) => "connection status", PeerMessage::BlockProposalNotification(_) => "block proposal notification", @@ -530,8 +525,6 @@ impl PeerMessage { PeerMessage::Transaction(_) => false, PeerMessage::TransactionNotification(_) => false, PeerMessage::TransactionRequest(_) => false, - PeerMessage::PeerListRequest => false, - PeerMessage::PeerListResponse(_) => false, PeerMessage::Bye => false, PeerMessage::ConnectionStatus(_) => false, PeerMessage::BlockProposalNotification(_) => false, @@ -558,8 +551,6 @@ impl PeerMessage { PeerMessage::Transaction(_) => true, PeerMessage::TransactionNotification(_) => false, PeerMessage::TransactionRequest(_) => false, - PeerMessage::PeerListRequest => false, - PeerMessage::PeerListResponse(_) => false, PeerMessage::Bye => false, PeerMessage::ConnectionStatus(_) => false, PeerMessage::BlockProposalNotification(_) => true, @@ -593,8 +584,6 @@ impl PeerMessage { PeerMessage::Transaction(_) => true, PeerMessage::TransactionNotification(_) => true, PeerMessage::TransactionRequest(_) => true, - PeerMessage::PeerListRequest => false, - PeerMessage::PeerListResponse(_) => false, PeerMessage::Bye => false, PeerMessage::ConnectionStatus(_) => false, PeerMessage::SyncCoverage(_) => true, From 7c91c9f80af8ea1d0ee26a8cda71397a18ead911 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Sun, 2 Aug 2026 04:52:15 +0300 Subject: [PATCH 2/5] feat(p2p): Remove deprecated id binding --- p2p/src/peer/peer_info.rs | 53 --------------------------------------- 1 file changed, 53 deletions(-) diff --git a/p2p/src/peer/peer_info.rs b/p2p/src/peer/peer_info.rs index ab994c0..84d47d9 100644 --- a/p2p/src/peer/peer_info.rs +++ b/p2p/src/peer/peer_info.rs @@ -1,68 +1,15 @@ use std::net::IpAddr; -use std::net::SocketAddr; use std::time::SystemTime; use libp2p::Multiaddr; -use libp2p::PeerId; use libp2p::multiaddr::Protocol; -use libp2p::multihash::Multihash; use serde::Deserialize; use serde::Serialize; -use sha2::Digest; -use sha2::Sha256; use super::InstanceId; use super::PeerStanding; use crate::peer::HandshakeData; -/// Derive a pseudorandom [`PeerId`] from a [`SocketAddr`]. -/// -/// This is a controversial feature, with accordingly complex motivation -- bear -/// with: -/// -/// This map is a compatibility layer bridging two architectures. With the -/// introduction of the libp2p network stack, all peer-related dictionaries -/// were modified to use the [`PeerId`] as the key instead of the -/// [`SocketAddr`]. Here is why the [`PeerId`] is a better choice: -/// - In the libp2p stack, the [`PeerId`] is cryptographically bound to the -/// peer's public key -- and the stack will refuse to connect if it fails to -/// verify that the peer really knows the matching secret key. -/// - The same peer (identified by public key) can have multiple -/// [`SocketAddr`]s, and even other internet-route-locators. While malicious -/// attackers can generate many key pairs, the point is that honest peers -/// will reuse the same public key but will be treated as different peers if -/// their [`SocketAddr`] is used as a stand-in for their identity. The -/// [`SocketAddr`] can change under benign circumstances, for instance if the -/// peer switches from WiFi to 4G, or it their ISP switches to a different -/// IP. So for honest peers, using the [`PeerId`] as the identifier leads to -/// less redundant work and greater peer set entropy. -/// - Malicious nodes must still use a variety of [`SocketAddr`]s if the goal -/// is to populate a victim's peer set with sybils and drive out honest -/// peers.Banning still happens at the level of [`SocketAddr`]s. So this -/// transition does not degrade security. -/// -/// However, since the legacy peer-to-peer stack has no concept of [`PeerId`], -/// it is difficult to access the dictionaries that now use the [`PeerId`] as -/// key, such as `peer_map` and `peer_standing`. This map deterministically -/// derives an identity ([`PeerId`]) from the peer's [`SocketAddr`]. This is a -/// controversial identification because a) there is no cryptographic -/// authentication or even tie to public keys; and b) leads to duplication of -/// work and weaker peer set entropy. Therefore, this map should only be used in -/// the context of the legacy peer-to-peer stack and, if the legacy peer-to-peer -/// stack is deprecated, this function should be deprecated along with it. -pub fn pseudorandom_peer_id(addr: &SocketAddr) -> PeerId { - let mut hasher = Sha256::new(); - hasher.update(b"legacy-mapping"); - hasher.update(addr.to_string().as_bytes()); - let hash_result = hasher.finalize(); - - // Use SHA2_256 (Code 0x12) which is the libp2p standard. - let mhash = Multihash::wrap(0x12, &hash_result) - .expect("SHA2-256 hash length is 32 bytes, which is valid for multihash"); - - PeerId::from_multihash(mhash).expect("SHA2-256 is the standard libp2p PeerId hash algorithm") -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] pub struct PeerConnectionInfo { listen_port: Option, From 5901709e190e18533813734bf15f19ca919a7375 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Sun, 2 Aug 2026 04:55:45 +0300 Subject: [PATCH 3/5] feat(node): Dont hold disconnection times anymore --- node/src/state/networking_state.rs | 20 ++------------------ 1 file changed, 2 insertions(+), 18 deletions(-) diff --git a/node/src/state/networking_state.rs b/node/src/state/networking_state.rs index e651195..a86e0c0 100644 --- a/node/src/state/networking_state.rs +++ b/node/src/state/networking_state.rs @@ -117,17 +117,6 @@ pub struct NetworkingState { /// sent from this client, or accepted from peers. /// Only the RPC server may update this flag. pub freeze: bool, - - /// Disconnection times of past peers. Can be used to determine if a connection - /// request should be accepted or rejected. - /// - /// Only records times of _graceful_ disconnections that were triggered by - /// _this_ node. That is, times of the following events are _not_ recorded: - /// - Graceful disconnect initiated by the peer. - /// - Abrupt disconnections, for example due to network failures. - /// - /// Only the peer tasks may update this map. - disconnection_times: HashMap, } impl NetworkingState { @@ -139,7 +128,6 @@ impl NetworkingState { sync_status: SyncStatus::Unknown, instance_id: rng().random(), freeze: false, - disconnection_times: HashMap::new(), } } @@ -239,11 +227,7 @@ impl NetworkingState { /// - Abrupt disconnections, for example due to network failures. /// /// Only the peer tasks may call this method. - pub(crate) fn register_peer_disconnection(&mut self, id: InstanceId, time: SystemTime) { - self.disconnection_times.insert(id, time); - } - - pub(crate) fn last_disconnection_time_of_peer(&self, id: InstanceId) -> Option { - self.disconnection_times.get(&id).copied() + pub(crate) fn register_peer_disconnection(&mut self, _id: InstanceId, _time: SystemTime) { + // Retired: we do not hold disconnection times anymore... Kept to see if we can do anything else w this later } } From 0034b55c876560618dd0a6681e45c4ee5d3aa995 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Sun, 2 Aug 2026 05:01:38 +0300 Subject: [PATCH 4/5] feat(node): Remove tokio-serde and PT2M dependency over main loop --- node/Cargo.toml | 1 - node/src/application/loops/main_loop.rs | 43 ------------------- .../application/loops/peer_loop/channel.rs | 2 - node/src/lib.rs | 4 +- 4 files changed, 1 insertion(+), 49 deletions(-) diff --git a/node/Cargo.toml b/node/Cargo.toml index cfb52de..8d73134 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -92,7 +92,6 @@ serde_json = "1.0" strum = { version = "0.27.0", features = ["derive"] } tasm-lib = "7.0.0" tokio = { version = "1.47", features = ["fs", "io-util", "macros", "process", "signal", "sync", "time", "tracing"] } -tokio-serde = { version = "0.8", features = ["bincode", "json"] } tokio-util = { version = "0.7", features = ["codec", "rt", "compat"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = [ diff --git a/node/src/application/loops/main_loop.rs b/node/src/application/loops/main_loop.rs index 259d327..8f2c9d8 100644 --- a/node/src/application/loops/main_loop.rs +++ b/node/src/application/loops/main_loop.rs @@ -1,29 +1,20 @@ pub(crate) mod network_event_handler; -use std::collections::HashMap; use std::collections::HashSet; use std::net::IpAddr; -use std::net::SocketAddr; - -use std::sync::Arc; use std::time::Duration; use std::time::SystemTime; use anyhow::Result; use itertools::Either; use itertools::Itertools; -use libp2p::PeerId; use nyks_consensus::block::Block; -use nyks_p2p::peer::handshake_data::HandshakeData; -use nyks_p2p::peer::peer_info::PeerInfo; use nyks_p2p::peer::transaction_notification::TransactionNotification; use rand::prelude::IteratorRandom; -use tokio::net::TcpListener; use tokio::select; use tokio::signal; use tokio::sync::broadcast; use tokio::sync::mpsc; -use tokio::sync::Semaphore; use tokio::task::JoinHandle; use tokio::time; use tokio::time::Instant; @@ -33,7 +24,6 @@ use tracing::error; use tracing::info; use tracing::warn; -use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; use crate::application::loops::channel::RPCServerToMain; use crate::application::loops::peer_loop::channel::MainToPeerTask; use crate::application::loops::peer_loop::channel::PeerTaskToMain; @@ -55,7 +45,6 @@ const PEER_DISCOVERY_INTERVAL: Duration = Duration::from_secs(2 * 60); const SYNC_REQUEST_INTERVAL: Duration = Duration::from_secs(3); const MEMPOOL_PRUNE_INTERVAL: Duration = Duration::from_secs(30 * 60); -const POTENTIAL_PEER_MAX_COUNT_AS_A_FACTOR_OF_MAX_PEERS: usize = 20; pub(crate) const MAX_NUM_DIGESTS_IN_BATCH_REQUEST: usize = 200; /// MainLoop is the immutable part of the input for the main loop function @@ -66,10 +55,6 @@ pub struct MainLoopHandler { // note: broadcast::Sender::send() does not block main_to_peer_broadcast_tx: broadcast::Sender, - // note: mpsc::Sender::send() blocks if channel full. - // locks should not be held across it. - peer_task_to_main_tx: mpsc::Sender, - network_command_tx: mpsc::Sender, peer_task_to_main_rx: mpsc::Receiver, @@ -107,7 +92,6 @@ impl MainLoopHandler { pub(crate) fn new( global_state_lock: GlobalStateLock, main_to_peer_broadcast_tx: broadcast::Sender, - peer_task_to_main_tx: mpsc::Sender, network_command_tx: mpsc::Sender, peer_task_to_main_rx: mpsc::Receiver, @@ -118,7 +102,6 @@ impl MainLoopHandler { Self { global_state_lock, main_to_peer_broadcast_tx, - peer_task_to_main_tx, network_command_tx, peer_task_to_main_rx, @@ -451,32 +434,6 @@ impl MainLoopHandler { let pmsg = MainToPeerTask::BlockProposalNotification((&*block).into()); self.main_to_peer_broadcast(pmsg); } - PeerTaskToMain::DisconnectFromLongestLivedPeer => { - let global_state = self.global_state_lock.lock_guard().await; - - // get all peers - let all_peers = global_state.net.peer_map.iter(); - - // filter out CLI peers - let cli_peers = global_state.cli().peers.iter().collect::>(); - let disconnect_candidates = all_peers - .filter(|(_id, info)| !cli_peers.contains(&info.address())) - .filter(|p| multiaddr_to_socketaddr(&p.1.address()).is_some()); - - // find the one with the oldest connection - let longest_lived_peer = - disconnect_candidates.min_by(|(_, peer_info_left), (_, peer_info_right)| { - peer_info_left - .connection_established() - .cmp(&peer_info_right.connection_established()) - }); - - // tell to disconnect - if let Some((peer_id, _peer_info)) = longest_lived_peer { - let pmsg = MainToPeerTask::Disconnect(*peer_id); - self.main_to_peer_broadcast(pmsg); - } - } PeerTaskToMain::NewSyncTarget(new_target) => { if let Some(sync_loop) = &mut main_loop_state.maybe_sync_loop { // Double-check new leadership status. Race condition. diff --git a/node/src/application/loops/peer_loop/channel.rs b/node/src/application/loops/peer_loop/channel.rs index 876d44b..2366487 100644 --- a/node/src/application/loops/peer_loop/channel.rs +++ b/node/src/application/loops/peer_loop/channel.rs @@ -99,7 +99,6 @@ pub(crate) enum PeerTaskToMain { Transaction(Box), BlockProposal(Box), - DisconnectFromLongestLivedPeer, NewSyncTarget(Box), NewSyncBlock(Box, PeerId), NewPeer(PeerId), @@ -128,7 +127,6 @@ impl PeerTaskToMain { PeerTaskToMain::AddPeerMaxBlockHeight { .. } => "add peer max block height", PeerTaskToMain::Transaction(_) => "transaction", PeerTaskToMain::BlockProposal(_) => "block proposal", - PeerTaskToMain::DisconnectFromLongestLivedPeer => "disconnect from longest lived peer", PeerTaskToMain::NewSyncTarget(_block) => "new sync target", PeerTaskToMain::NewSyncBlock(_block, _socket_addr) => "new sync block", PeerTaskToMain::NewPeer { .. } => "new peer", diff --git a/node/src/lib.rs b/node/src/lib.rs index 0745771..750735e 100644 --- a/node/src/lib.rs +++ b/node/src/lib.rs @@ -30,7 +30,6 @@ pub mod util_types; use std::env; use std::path::PathBuf; -use anyhow::Context; use anyhow::Result; use application::config::cli_args; use chrono::DateTime; @@ -52,7 +51,6 @@ use tracing::warn; use crate::application::config::data_directory::DataDirectory; use crate::application::config::identity::resolve_identity; -use crate::application::config::parser::multiaddr::multiaddr_to_socketaddr; use crate::application::loops::channel::RPCServerToMain; use crate::application::loops::main_loop::MainLoopHandler; use crate::application::loops::peer_loop::channel::MainToPeerTask; @@ -102,6 +100,7 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { broadcast::channel::(PEER_CHANNEL_CAPACITY); // Add the MPSC (multi-producer, single consumer) channel for peer-task-to-main communication + // TODO: Think about other use cases and if theres none optimize/cleanup let (peer_task_to_main_tx, peer_task_to_main_rx) = mpsc::channel::(PEER_CHANNEL_CAPACITY); @@ -218,7 +217,6 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { Ok(MainLoopHandler::new( global_state_lock, main_to_peer_broadcast_tx, - peer_task_to_main_tx, network_command_tx, peer_task_to_main_rx, rpc_server_to_main_rx, From a5b54eeeb3538c87c918c71681379ad1dd058997 Mon Sep 17 00:00:00 2001 From: KaffinPX Date: Mon, 3 Aug 2026 05:30:21 +0300 Subject: [PATCH 5/5] feat(node): More cleanup and fix StickyPeers stuck on dialing --- consensus/src/block/pow.rs | 1 - node/src/application/loops/peer_loop.rs | 5 ---- node/src/application/network/actor.rs | 33 +++++++++++++++++++------ node/src/application/network/codec.rs | 2 +- node/src/application/network/mod.rs | 2 +- node/src/lib.rs | 6 +---- p2p/src/peer.rs | 1 - 7 files changed, 29 insertions(+), 21 deletions(-) diff --git a/consensus/src/block/pow.rs b/consensus/src/block/pow.rs index f17331d..9d56ac4 100644 --- a/consensus/src/block/pow.rs +++ b/consensus/src/block/pow.rs @@ -23,7 +23,6 @@ use crate::BFieldElement; use crate::block::Block; use crate::block::block_header::BlockHeader; use crate::block::block_kernel::BlockKernel; -use crate::consensus_rule_set::ConsensusRuleSet; use crate::proof_abstractions::mast_hash::MastHash; /// Determines the number of leafs in the Merkle tree in the guesser buffer. diff --git a/node/src/application/loops/peer_loop.rs b/node/src/application/loops/peer_loop.rs index 42442c2..32a3216 100644 --- a/node/src/application/loops/peer_loop.rs +++ b/node/src/application/loops/peer_loop.rs @@ -106,7 +106,6 @@ pub struct PeerLoopHandler { peer_address: Multiaddr, peer_handshake_data: HandshakeData, inbound_connection: bool, - distance: u8, rng: StdRng, #[cfg(test)] mock_now: Option, @@ -120,7 +119,6 @@ impl PeerLoopHandler { peer_address: Multiaddr, peer_handshake_data: HandshakeData, inbound_connection: bool, - distance: u8, ) -> Self { Self { to_main_tx, @@ -129,7 +127,6 @@ impl PeerLoopHandler { peer_address, peer_handshake_data, inbound_connection, - distance, rng: StdRng::from_rng(&mut rand::rng()), #[cfg(test)] mock_now: None, @@ -146,7 +143,6 @@ impl PeerLoopHandler { peer_address: Multiaddr, peer_handshake_data: HandshakeData, inbound_connection: bool, - distance: u8, mocked_time: Timestamp, ) -> Self { Self { @@ -156,7 +152,6 @@ impl PeerLoopHandler { peer_address, peer_handshake_data, inbound_connection, - distance, mock_now: Some(mocked_time), rng: StdRng::from_rng(&mut rand::rng()), } diff --git a/node/src/application/network/actor.rs b/node/src/application/network/actor.rs index 3de010f..79f6c63 100644 --- a/node/src/application/network/actor.rs +++ b/node/src/application/network/actor.rs @@ -968,6 +968,32 @@ impl NetworkActor { connection_id, .. } => { + let address = endpoint.get_remote_address().clone(); + + // Update sticky-peer state unconditionally: even if this connection + // turns out to be a duplicate (and gets ignored below), it still + // proves the peer at this Multiaddr is alive and reachable. If we + // don't do this here, a sticky peer's state can get stuck in + // `Dialing` forever whenever a race lets some other connection to the + // same peer_id land first (e.g. an inbound dial, or a Kademlia-driven + // connection), even though we are clearly talking to it. + if self.sticky_peers.contains_key(&address) { + self.sticky_peers.entry(address.clone()).and_modify(|p| { + match p { + StickyPeer::None | StickyPeer::Dialing(_) => { + tracing::debug!(%peer_id, "Found peer id of sticky peer {address}."); + *p = StickyPeer::Connected(peer_id); + } + StickyPeer::Connected(pid) => { + if *pid != peer_id { + tracing::debug!(%peer_id, "Found *new* peer id of sticky peer {address}."); + *pid = peer_id; + } + } + } + }); + } + // Gatekeep: one connection per peer. // We allow the duplicate connection to remain because libp2p // deals with that and we do not want to interfere with that @@ -980,8 +1006,6 @@ impl NetworkActor { return Ok(()); } - let address = endpoint.get_remote_address().clone(); - // Check for banned IPs again. The catch above in // `IncomingConnection` is a good first filter but because of // race conditions, masked addresses, and automatic outgoing @@ -2123,10 +2147,6 @@ impl NetworkActor { raw_stream: libp2p::Stream, from_main_rx: tokio::sync::broadcast::Receiver, ) -> Option> { - // Counts the number of hops between the node and peers it is connected - // to. We probably don't need this for the libp2p wrapper. - const DISTANCE_TO_CONNECTED_PEER: u8 = 1u8; - // Keep track of which peers get upgraded connections. Prevent same // peer from getting upgraded multiple times. let num_upgraded_peers = { @@ -2159,7 +2179,6 @@ impl NetworkActor { peer_address, remote_handshake, rand::rng().random_bool(0.5f64), - DISTANCE_TO_CONNECTED_PEER, ); let peer_stream = bridge_libp2p_stream(raw_stream); diff --git a/node/src/application/network/codec.rs b/node/src/application/network/codec.rs index d38401a..6762c2c 100644 --- a/node/src/application/network/codec.rs +++ b/node/src/application/network/codec.rs @@ -11,4 +11,4 @@ pub(crate) fn get_codec_rules() -> LengthDelimitedCodec { let mut codec_rules = LengthDelimitedCodec::new(); codec_rules.set_max_frame_length(MAX_PEER_FRAME_LENGTH_IN_BYTES); codec_rules -} \ No newline at end of file +} diff --git a/node/src/application/network/mod.rs b/node/src/application/network/mod.rs index 46c9964..2edda12 100644 --- a/node/src/application/network/mod.rs +++ b/node/src/application/network/mod.rs @@ -3,13 +3,13 @@ pub(crate) mod address_book; pub(crate) mod ban; pub(crate) mod bridge; pub(crate) mod channel; +pub(crate) mod codec; pub(crate) mod config; pub(crate) mod gateway; pub(crate) mod handshake; pub mod overview; pub(crate) mod reachability; pub(crate) mod stack; -pub(crate) mod codec; #[cfg(any(test, feature = "arbitrary-impls"))] pub mod arbitrary; diff --git a/node/src/lib.rs b/node/src/lib.rs index 750735e..7c9e3e3 100644 --- a/node/src/lib.rs +++ b/node/src/lib.rs @@ -132,11 +132,7 @@ pub async fn initialize(cli_args: cli_args::Args) -> Result { // Set up the libp2p NetworkActor info!("Setting up Network Actor"); - let cli_peers_for_network_actor = cli_args - .peers - .iter() - .cloned() - .collect_vec(); + let cli_peers_for_network_actor = cli_args.peers.iter().cloned().collect_vec(); let network_config = NetworkConfig::default() .with_subdirectory(data_directory.network_subdirectory()) .with_network(cli_args.network) diff --git a/p2p/src/peer.rs b/p2p/src/peer.rs index 32db628..033b58a 100644 --- a/p2p/src/peer.rs +++ b/p2p/src/peer.rs @@ -7,7 +7,6 @@ pub mod transfer_block; pub mod transfer_transaction; use std::fmt::Display; -use std::net::SocketAddr; use std::time::SystemTime; use handshake_data::HandshakeData;