From 81f8f6e7110139c0cde7b97f4ee703900b9f88b5 Mon Sep 17 00:00:00 2001 From: viraladmin <00purple@gmail.com> Date: Sat, 11 Jul 2026 14:42:46 -0600 Subject: [PATCH] bootstrap process changes --- src/records/memory/network_mapping/add.rs | 49 ++++++++++---- src/records/memory/network_mapping/mod.rs | 3 + .../memory/network_mapping/persistence.rs | 20 ++++-- src/records/memory/network_mapping/queries.rs | 15 ++++- src/records/memory/network_mapping/structs.rs | 5 +- src/rpc/client/handshake.rs | 65 ++++++++++++++++++- src/rpc/client/handshake_processing.rs | 44 +++---------- src/rpc/client/structs.rs | 1 + src/rpc/command_maps.rs | 1 - src/rpc/commands/add_network_node.rs | 5 +- src/rpc/commands/mod.rs | 1 - src/rpc/commands/random_node.rs | 26 +++++--- src/rpc/server/rpc_command_loop.rs | 23 +++---- src/startup/network_broadcast.rs | 18 ++++- 14 files changed, 193 insertions(+), 83 deletions(-) diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index b8a4d36..fe91815 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -37,8 +37,8 @@ impl NodeInfo { } let data = format!( - "{}{}{}{}", - edit.address, edit.ip, edit.modified_by, edit.modified_timestamp + "{}{}{}{}{}", + edit.address, edit.ip, edit.port, edit.modified_by, edit.modified_timestamp ); let hashed_data = skein_256_hash_data(&data); let pubkey = crate::records::wallet_registry::resolve_pubkey_from_short_address( @@ -62,6 +62,7 @@ impl NodeInfo { if let Some(existing_node) = address_map.get_mut(&edit.address) { if existing_node.deleted_timestamp > 0 { existing_node.ip = edit.ip; + existing_node.port = edit.port; existing_node.added_by = edit.modified_by; existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature; @@ -75,7 +76,14 @@ impl NodeInfo { if existing_node.ip != edit.ip { return Err("active node must be deleted before changing IP".to_string()); } - if edit.modified_timestamp < existing_node.added_timestamp { + if existing_node.port != edit.port + && edit.modified_timestamp >= existing_node.added_timestamp + { + existing_node.port = edit.port; + existing_node.added_by = edit.modified_by; + existing_node.added_timestamp = edit.modified_timestamp; + existing_node.added_signature = edit.modified_signature; + } else if edit.modified_timestamp < existing_node.added_timestamp { existing_node.added_by = edit.modified_by; existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature; @@ -96,6 +104,7 @@ impl NodeInfo { address_map.insert(edit.address, { let mut node = NodeInfo::new( edit.ip, + edit.port, 0, edit.modified_by, edit.modified_timestamp, @@ -124,6 +133,7 @@ impl NodeInfo { SignedNodeEdit { address: address.to_string(), ip: node.ip.clone(), + port: node.port, modified_by: node.added_by.clone(), modified_timestamp: node.added_timestamp, modified_signature: node.added_signature.clone(), @@ -143,6 +153,7 @@ impl NodeInfo { // skipping the source peer that already sent the update. let message_type = edittype.message_type(); let ip_bytes = ip_to_binary(&edit.ip); + let port_bytes = edit.port.to_le_bytes(); let address_bytes = match Wallet::short_address_to_bytes(&edit.address) { Some(bytes) => bytes, None => { @@ -200,6 +211,7 @@ impl NodeInfo { message.extend_from_slice(&hashmap_key); message.extend_from_slice(&address_bytes); message.extend_from_slice(&ip_bytes); + message.extend_from_slice(&port_bytes); message.extend_from_slice(&modified_by_bytes); message.extend_from_slice(&modified_timestamp_bytes); message.extend_from_slice(&modified_signature_bytes); @@ -248,11 +260,14 @@ impl NodeInfo { if let Some(existing_node) = address_map.get(&edit.address) { if existing_node.deleted_timestamp == 0 { if existing_node.ip == edit.ip { - return RpcResponse::Binary(b"Success".to_vec()); + if existing_node.port == edit.port { + return RpcResponse::Binary(b"Success".to_vec()); + } + } else { + return RpcResponse::Binary( + b"Error: Active node must be deleted before changing IP".to_vec(), + ); } - return RpcResponse::Binary( - b"Error: Active node must be deleted before changing IP".to_vec(), - ); } } drop(address_map); @@ -262,13 +277,19 @@ impl NodeInfo { // records remain canonical and are not re-signed on reconnect. edit.modified_timestamp = current_timestamp; edit.modified_by = wallet.saved.short_address.clone(); - edit.modified_signature = - Self::added_signature(&edit.address, &edit.ip, current_timestamp, &wallet).await; + edit.modified_signature = Self::added_signature( + &edit.address, + &edit.ip, + edit.port, + current_timestamp, + &wallet, + ) + .await; } let data = format!( - "{}{}{}{}", - edit.address, edit.ip, edit.modified_by, edit.modified_timestamp + "{}{}{}{}{}", + edit.address, edit.ip, edit.port, edit.modified_by, edit.modified_timestamp ); let hashed_data = skein_256_hash_data(&data); @@ -359,6 +380,7 @@ impl NodeInfo { if let Some(existing_node) = address_map.get_mut(&edit.address) { if existing_node.deleted_timestamp > 0 { existing_node.ip = edit.ip.clone(); + existing_node.port = edit.port; existing_node.added_by = edit.modified_by.clone(); existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature.clone(); @@ -371,6 +393,10 @@ impl NodeInfo { b"Error: Active node must be deleted before changing IP".to_vec(), ); } + if existing_node.port != edit.port { + existing_node.port = edit.port; + state_changed = true; + } if edit.modified_timestamp < existing_node.added_timestamp { existing_node.added_by = edit.modified_by.clone(); existing_node.added_timestamp = edit.modified_timestamp; @@ -398,6 +424,7 @@ impl NodeInfo { address_map.insert(edit.address.clone(), { let mut node = NodeInfo::new( edit.ip.clone(), + edit.port, 0, edit.modified_by.clone(), edit.modified_timestamp, diff --git a/src/records/memory/network_mapping/mod.rs b/src/records/memory/network_mapping/mod.rs index 3f42af6..a38414d 100644 --- a/src/records/memory/network_mapping/mod.rs +++ b/src/records/memory/network_mapping/mod.rs @@ -36,6 +36,7 @@ static NODE_RUNTIME_STARTED_MILLIS: OnceLock = OnceLock::new(); #[derive(Debug)] pub struct NodeInfo { ip: String, + port: u16, blocks_mined: u8, added_by: String, added_timestamp: u64, @@ -68,6 +69,7 @@ impl NodeInfo { fn new( ip: String, + port: u16, blocks_mined: u8, added_by: String, added_timestamp: u64, @@ -75,6 +77,7 @@ impl NodeInfo { ) -> Self { NodeInfo { ip, + port, blocks_mined, added_by, added_timestamp, diff --git a/src/records/memory/network_mapping/persistence.rs b/src/records/memory/network_mapping/persistence.rs index 215da85..3e4e51c 100644 --- a/src/records/memory/network_mapping/persistence.rs +++ b/src/records/memory/network_mapping/persistence.rs @@ -4,7 +4,7 @@ use crate::common::network_paths_and_settings::block_extension_and_paths; use crate::records::memory::network_mapping::structs::{ NODE_ADDED_BY_OFFSET, NODE_ADDED_SIGNATURE_OFFSET, NODE_ADDED_TIMESTAMP_OFFSET, NODE_BLOCKS_MINED_OFFSET, NODE_DELETED_BLOCK_OFFSET, NODE_DELETED_TIMESTAMP_OFFSET, - NODE_IP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_RECORD_FIXED_BYTES, + NODE_IP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_PORT_OFFSET, NODE_RECORD_FIXED_BYTES, }; use crate::{create_dir_all, read, PathBuf}; @@ -41,6 +41,7 @@ impl NodeInfo { data.extend_from_slice(&address_bytes); data.extend_from_slice(&ip_to_binary(&node_info.ip)); + data.extend_from_slice(&node_info.port.to_le_bytes()); data.push(node_info.blocks_mined); if added_by_bytes.len() == Wallet::SHORT_ADDRESS_BYTES_LENGTH { data.extend_from_slice(&added_by_bytes); @@ -87,7 +88,12 @@ impl NodeInfo { let Some(address) = Wallet::bytes_to_short_address(&chunk[0..NODE_IP_OFFSET]) else { continue; }; - let ip = binary_to_ip(chunk[NODE_IP_OFFSET..NODE_BLOCKS_MINED_OFFSET].to_vec()); + let ip = binary_to_ip(chunk[NODE_IP_OFFSET..NODE_PORT_OFFSET].to_vec()); + let port = u16::from_le_bytes( + chunk[NODE_PORT_OFFSET..NODE_BLOCKS_MINED_OFFSET] + .try_into() + .unwrap(), + ); let blocks_mined = chunk[NODE_BLOCKS_MINED_OFFSET]; let added_by_bytes = &chunk[NODE_ADDED_BY_OFFSET..NODE_ADDED_TIMESTAMP_OFFSET]; let added_by = if added_by_bytes.iter().all(|&byte| byte == 0) { @@ -113,8 +119,14 @@ impl NodeInfo { .unwrap(), ); - let mut node = - NodeInfo::new(ip, blocks_mined, added_by, added_timestamp, added_signature); + let mut node = NodeInfo::new( + ip, + port, + blocks_mined, + added_by, + added_timestamp, + added_signature, + ); node.deleted_timestamp = deleted_timestamp; node.deleted_block = deleted_block; node.monitoring = Vec::new(); diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index a0522b5..995a11c 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -38,6 +38,16 @@ impl NodeInfo { .collect() } + pub async fn active_node_endpoints() -> Vec { + let map = ADDRESS_MAP.lock().await; + map.values() + // Outgoing connection fill needs a reachable endpoint, but node + // uniqueness still remains IP-based elsewhere in the map. + .filter(|node_info| node_info.deleted_timestamp == 0) + .map(|node_info| format!("{}:{}", node_info.ip, node_info.port)) + .collect() + } + pub async fn eligible_sponsor_ips() -> Vec { let current_timestamp = Utc::now().timestamp_millis() as u64; let map = ADDRESS_MAP.lock().await; @@ -79,6 +89,7 @@ impl NodeInfo { None => continue, }; let ip_bytes = ip_to_binary(&node_info.ip); + let port_bytes = node_info.port.to_le_bytes(); let blocks_mined = node_info.blocks_mined; let added_by_bytes = Wallet::short_address_to_bytes(&node_info.added_by).unwrap_or_default(); @@ -95,6 +106,7 @@ impl NodeInfo { // synchronization. data.extend_from_slice(&address_bytes); data.extend_from_slice(&ip_bytes); + data.extend_from_slice(&port_bytes); data.push(blocks_mined); if added_by_bytes.len() == Wallet::SHORT_ADDRESS_BYTES_LENGTH { data.extend_from_slice(&added_by_bytes); @@ -120,6 +132,7 @@ impl NodeInfo { pub async fn added_signature( address: &str, ip: &str, + port: u16, current_timestamp: u64, wallet: &Arc, ) -> String { @@ -128,7 +141,7 @@ impl NodeInfo { let added_by = wallet.saved.short_address.clone(); let private_key = &wallet.saved.private_key; - let data = format!("{address}{ip}{added_by}{current_timestamp}"); + let data = format!("{address}{ip}{port}{added_by}{current_timestamp}"); let hashed_data = skein_256_hash_data(&data); Wallet::sign_transaction(&hashed_data, private_key).await } diff --git a/src/records/memory/network_mapping/structs.rs b/src/records/memory/network_mapping/structs.rs index 2d8a149..038df76 100644 --- a/src/records/memory/network_mapping/structs.rs +++ b/src/records/memory/network_mapping/structs.rs @@ -5,6 +5,7 @@ use crate::Arc; use crate::Mutex; pub const NODE_IP_BYTES: usize = 16; +pub const NODE_PORT_BYTES: usize = 2; pub const NODE_BLOCKS_MINED_BYTES: usize = 1; pub const NODE_TIMESTAMP_BYTES: usize = 8; pub const NODE_DELETED_BLOCK_BYTES: usize = 4; @@ -12,7 +13,8 @@ pub const NODE_MONITOR_COUNT_BYTES: usize = 2; pub const NODE_ADDRESS_OFFSET: usize = 0; pub const NODE_IP_OFFSET: usize = NODE_ADDRESS_OFFSET + Wallet::SHORT_ADDRESS_BYTES_LENGTH; -pub const NODE_BLOCKS_MINED_OFFSET: usize = NODE_IP_OFFSET + NODE_IP_BYTES; +pub const NODE_PORT_OFFSET: usize = NODE_IP_OFFSET + NODE_IP_BYTES; +pub const NODE_BLOCKS_MINED_OFFSET: usize = NODE_PORT_OFFSET + NODE_PORT_BYTES; pub const NODE_ADDED_BY_OFFSET: usize = NODE_BLOCKS_MINED_OFFSET + NODE_BLOCKS_MINED_BYTES; pub const NODE_ADDED_TIMESTAMP_OFFSET: usize = NODE_ADDED_BY_OFFSET + Wallet::SHORT_ADDRESS_BYTES_LENGTH; @@ -28,6 +30,7 @@ pub const NODE_RECORD_FIXED_BYTES: usize = NODE_MONITOR_COUNT_OFFSET + NODE_MONI pub struct SignedNodeEdit { pub address: String, pub ip: String, + pub port: u16, pub modified_by: String, pub modified_timestamp: u64, pub modified_signature: String, diff --git a/src/rpc/client/handshake.rs b/src/rpc/client/handshake.rs index 488bc9d..9aaf307 100644 --- a/src/rpc/client/handshake.rs +++ b/src/rpc/client/handshake.rs @@ -1,5 +1,6 @@ use crate::common::network_startup::get_listen_ip; use crate::io; +use crate::log::info; use crate::rpc::client::handshake_message::prepare_handshake_message; use crate::rpc::client::handshake_processing::process_handshake_response; use crate::rpc::client::structs::{Connect, Handshake}; @@ -8,13 +9,60 @@ use crate::rpc::handshake_constants::HANDSHAKE_RESPONSE_BYTES; use crate::IpAddr; use crate::SocketAddr; use crate::TcpStream; -use crate::{timeout, AsyncReadExt, AsyncWriteExt, Duration}; +use crate::{sleep, timeout, AsyncReadExt, AsyncWriteExt, Duration}; use tokio::net::TcpSocket; const OUTBOUND_CONNECT_TIMEOUT_SECONDS: u64 = 2; const HANDSHAKE_RESPONSE_TIMEOUT_SECONDS: u64 = 2; -pub async fn connect_and_handshake(params: Connect) -> Result<(), Box> { +pub async fn connect_and_handshake( + params: Connect, +) -> Result<(), Box> { + let original_params = params.clone(); + let err = match connect_and_handshake_once(params).await { + Ok(()) => return Ok(()), + Err(err) => err, + }; + + let original_error = err.to_string(); + let Some(sponsor_endpoint) = sponsor_endpoint_from_error(&original_error) else { + return Err(err); + }; + let sponsor_addr: SocketAddr = match sponsor_endpoint.parse() { + Ok(addr) => addr, + Err(_) => return Err(err), + }; + if sponsor_addr == original_params.addr { + return Err(err); + } + drop(err); + + info!( + "[handshake] peer {} rejected network add; trying sponsor {}", + original_params.node_ip, sponsor_endpoint + ); + sleep(Duration::from_secs(2)).await; + + let sponsor_params = Connect { + addr: sponsor_addr, + node_ip: sponsor_endpoint.clone(), + db: original_params.db, + wallet: original_params.wallet, + map: original_params.map, + first: original_params.first, + }; + connect_and_handshake_once(sponsor_params) + .await + .map_err(|sponsor_err| { + Box::new(io::Error::other(format!( + "Sponsor redirect failed: original={original_error}; sponsor={sponsor_endpoint}; sponsor_error={sponsor_err}" + ))) as Box + }) +} + +async fn connect_and_handshake_once( + params: Connect, +) -> Result<(), Box> { // Outbound node sockets bind to the configured local IP before // connecting so the socket source matches the advertised node IP. let stream = match connect_from_configured_ip(params.addr).await { @@ -36,7 +84,18 @@ pub async fn connect_and_handshake(params: Connect) -> Result<(), Box) + .map_err(|err| Box::new(err) as Box) +} + +fn sponsor_endpoint_from_error(error: &str) -> Option { + let (_, sponsor) = error.split_once("sponsor=")?; + let endpoint = sponsor + .split_whitespace() + .next() + .unwrap_or_default() + .trim_matches(|ch| ch == '"' || ch == '\'' || ch == ',' || ch == ';'); + endpoint.parse::().ok()?; + Some(endpoint.to_string()) } async fn connect_from_configured_ip(remote_addr: SocketAddr) -> io::Result { diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 33953db..adf3653 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -1,4 +1,3 @@ -use crate::common::binary_conversions::binary_to_ip_port; use crate::common::check_genesis::genesis_checkup; use crate::common::network_startup::get_ip_and_port; use crate::common::skein::skein_256_hash_data; @@ -19,7 +18,8 @@ use crate::records::memory::connections::{ set_reconnect_context, startup_synced_peer_streams, CONNECTIONS, }; use crate::records::memory::enums::{ClientType, ConnectionType}; -use crate::records::memory::response_channels::{reserve_entry, Command}; +use crate::records::memory::network_mapping::NodeInfo; +use crate::records::memory::response_channels::Command; use crate::records::memory::structs::{Connection, StoreConnectionParams}; use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::handshake::connect_and_handshake; @@ -27,23 +27,22 @@ use crate::rpc::client::register_wallet::register_connected_wallet; use crate::rpc::client::structs::{Connect, Handshake}; use crate::rpc::client::syncing::node_syncing; use crate::rpc::client::wallet_registry_sync::sync_wallet_registry_with_retries; -use crate::rpc::command_maps::RPC_RANDOM_NODE; use crate::rpc::handshake_constants::{ HANDSHAKE_ADDRESS_OFFSET, HANDSHAKE_MESSAGE_BYTES, HANDSHAKE_RESPONSE_BYTES, HANDSHAKE_SIGNATURE_OFFSET, }; -use crate::rpc::responses::RpcResponse; use crate::rpc::server::connection_memory_manager::remove_key_from_memory; use crate::rpc::server::rpc_command_loop::start_loop; use crate::sled::Db; use crate::sleep; use crate::startup::network_broadcast::announce_self_to_network; use crate::startup::remote_height::request_remote_height; -use crate::timeout; +use crate::thread_rng; use crate::wallets::structures::Wallet; use crate::Arc; use crate::Duration; use crate::Mutex; +use crate::SliceRandom; use crate::SocketAddr; use crate::TcpStream; @@ -79,12 +78,13 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(), } let (_, _, local_endpoint) = get_ip_and_port().await; let max = SETTINGS.outgoing_connections; - let mut no_progress_count = 0; - let max_no_progress = 3; let mut current_key = params.connections_key.clone(); let mut stream = params.stream; params.first = false; - while no_progress_count < max_no_progress { + let mut candidate_endpoints = NodeInfo::active_node_endpoints().await; + candidate_endpoints.shuffle(&mut thread_rng()); + + for addr_string in candidate_endpoints { let outgoing_connections = { let connections = CONNECTIONS.read().await; connections @@ -96,28 +96,7 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(), break; } - let (hashmap_key, _tx, rx) = reserve_entry(params.map.clone()).await; - - let mut payload = vec![RPC_RANDOM_NODE]; - payload.extend_from_slice(&hashmap_key); - - RpcResponse::send_raw(&stream, Some(¤t_key), &payload).await; - - let mut rx = rx.lock().await; - let buffer = match timeout(Duration::from_secs(15), rx.recv()).await { - Ok(Some(buf)) => buf, - Ok(None) => return Err("Peer discovery channel closed".into()), - Err(_) => return Err("Timeout waiting for peer discovery response".into()), - }; - - if buffer.len() != 18 { - no_progress_count += 1; - continue; - } - - let addr_string = binary_to_ip_port(&buffer[0..18]); if addr_string == local_endpoint { - no_progress_count += 1; continue; } @@ -125,15 +104,13 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(), .await .is_some() { - no_progress_count += 1; continue; } let socket_addr: SocketAddr = match addr_string.parse() { Ok(addr) => addr, Err(_) => { - warn!("Invalid discovered peer from {current_key}: {addr_string}"); - no_progress_count += 1; + warn!("Invalid mapped peer endpoint: {addr_string}"); continue; } }; @@ -151,17 +128,14 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(), if let Err(err) = connect_and_handshake(connect).await { warn!("Failed to connect to discovered peer {addr_string}: {err}"); - no_progress_count += 1; continue; } if let Some(new_stream) = Connection::get_stream_from_memory(&addr_string).await { stream = new_stream; current_key = addr_string; - no_progress_count = 0; } else { warn!("Failed to retrieve new stream for: {addr_string}"); - no_progress_count += 1; } } diff --git a/src/rpc/client/structs.rs b/src/rpc/client/structs.rs index 00e7821..9edeba3 100644 --- a/src/rpc/client/structs.rs +++ b/src/rpc/client/structs.rs @@ -7,6 +7,7 @@ use crate::SocketAddr; use crate::TcpStream; // Connect carries the data needed to open a client connection and begin the handshake flow. +#[derive(Clone)] pub struct Connect { pub addr: SocketAddr, pub db: Db, diff --git a/src/rpc/command_maps.rs b/src/rpc/command_maps.rs index 8edaa38..1364ea2 100644 --- a/src/rpc/command_maps.rs +++ b/src/rpc/command_maps.rs @@ -18,7 +18,6 @@ use crate::blocks::vanity::VanityAddressTransaction; pub const RPC_NETWORK_INFO: u8 = 1; pub const RPC_BLOCK_HEIGHT: u8 = 2; -pub const RPC_RANDOM_NODE: u8 = 3; pub const RPC_TIME: u8 = 4; pub const RPC_DIFFICULTY: u8 = 5; pub const RPC_VALIDATE_TORRENT: u8 = 6; diff --git a/src/rpc/commands/add_network_node.rs b/src/rpc/commands/add_network_node.rs index 86df7f2..57b8555 100644 --- a/src/rpc/commands/add_network_node.rs +++ b/src/rpc/commands/add_network_node.rs @@ -17,7 +17,7 @@ pub async fn add_network_node( map: Arc>, ) -> Result<(u32, RpcResponse), String> { // Command 28 carries the signed node-add payload directly after the - // UID: node address, node IP, signer, timestamp, and signature. + // UID: node address, node IP, node port, signer, timestamp, and signature. let (uid, _) = read_bytes_from_stream::read_uid_from_stream(connections_key, stream_locked.clone()) .await?; @@ -30,6 +30,8 @@ pub async fn add_network_node( .ok_or_else(|| "error: Invalid short address bytes".to_string())?; let ip = read_bytes_from_stream::read_ip_from_stream(connections_key, stream_locked.clone()).await?; + let port = read_bytes_from_stream::read_u16_from_stream(connections_key, stream_locked.clone()) + .await?; let added_by_bytes = read_bytes_from_stream::read_usize_from_stream( connections_key, Wallet::SHORT_ADDRESS_BYTES_LENGTH, @@ -70,6 +72,7 @@ pub async fn add_network_node( edit: SignedNodeEdit { address, ip, + port, modified_by: added_by, modified_timestamp: added_timestamp, modified_signature: added_signature, diff --git a/src/rpc/commands/mod.rs b/src/rpc/commands/mod.rs index ada8809..56635af 100644 --- a/src/rpc/commands/mod.rs +++ b/src/rpc/commands/mod.rs @@ -23,7 +23,6 @@ pub mod network_monitor_add; pub mod network_monitor_remove; pub mod nft_list; pub mod nft_lookup; -pub mod random_node; pub mod receive_torrent; pub mod request_valid_nodes; pub mod route_reply; diff --git a/src/rpc/commands/random_node.rs b/src/rpc/commands/random_node.rs index 73a241d..7b1b1e9 100644 --- a/src/rpc/commands/random_node.rs +++ b/src/rpc/commands/random_node.rs @@ -1,10 +1,11 @@ +use crate::common::binary_conversions::binary_to_ip; use crate::records::block_height::get_block_height::get_height; use crate::records::memory::connections::CONNECTIONS; use crate::records::memory::network_mapping::NodeInfo; use crate::rpc::responses::RpcResponse; use crate::sled::Db; -pub async fn request_node(db: &Db, connections_key: &str) -> RpcResponse { +pub async fn eligible_sponsor_endpoint(db: &Db, connections_key: &str) -> Option { // Before the mature-network gate, node adds are intentionally permissive. // After height 10,000, only mature sponsors can be returned. let eligible_ips = if get_height(db) > 10000 { @@ -13,22 +14,31 @@ pub async fn request_node(db: &Db, connections_key: &str) -> RpcResponse { NodeInfo::active_node_ips().await }; if eligible_ips.is_empty() { - return RpcResponse::Binary(b"Error: No Eligible Sponsor Found".to_vec()); + return None; } - let result = { + { let connections_lock = CONNECTIONS.read().await; if let Some(connection) = connections_lock.as_ref() { connection.get_random_connection(Some(connections_key), Some(&eligible_ips)) } else { None } - }; + } + .map(|(ip, port)| format!("{}:{}", binary_to_ip(ip), port)) +} - match result { - Some((mut ip, port)) => { - let port_bytes = port.to_le_bytes(); - ip.extend(port_bytes); +pub async fn request_node(db: &Db, connections_key: &str) -> RpcResponse { + match eligible_sponsor_endpoint(db, connections_key).await { + Some(endpoint) => { + let Some((ip, port)) = endpoint.rsplit_once(':') else { + return RpcResponse::Binary(b"Error: No Eligible Sponsor Found".to_vec()); + }; + let Ok(port) = port.parse::() else { + return RpcResponse::Binary(b"Error: No Eligible Sponsor Found".to_vec()); + }; + let mut ip = crate::common::binary_conversions::ip_to_binary(ip); + ip.extend(port.to_le_bytes()); RpcResponse::Binary(ip) } None => { diff --git a/src/rpc/server/rpc_command_loop.rs b/src/rpc/server/rpc_command_loop.rs index cf2a2be..6926316 100644 --- a/src/rpc/server/rpc_command_loop.rs +++ b/src/rpc/server/rpc_command_loop.rs @@ -4,7 +4,9 @@ use crate::log::warn; use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::Command; use crate::rpc::server::command_loop_state::next_incoming_command; -use crate::rpc::server::connection_memory_manager::remove_stream_from_memory; +use crate::rpc::server::connection_memory_manager::{ + remove_key_from_memory, remove_stream_from_memory, +}; use crate::rpc::*; use crate::sled::Db; use crate::wallets::structures::Wallet; @@ -60,19 +62,6 @@ pub async fn start_loop( .send(&stream_locked, Some(&connections_key), uid) .await; } - 3 => { - // request a random node - let (uid, _) = read_bytes_from_stream::read_uid_from_stream( - &connections_key, - stream_locked.clone(), - ) - .await?; - - let result = commands::random_node::request_node(&db, &connections_key).await; - result - .send(&stream_locked, Some(&connections_key), uid) - .await; - } 4 => { // request the current time let (uid, _) = read_bytes_from_stream::read_uid_from_stream( @@ -563,9 +552,15 @@ pub async fn start_loop( map.clone(), ) .await?; + let should_drop_rejected_miner = client_type == ClientType::Miner + && matches!(&result, responses::RpcResponse::Binary(bytes) if String::from_utf8_lossy(bytes).starts_with("Error:")); result .send(&stream_locked, Some(&connections_key), uid) .await; + if should_drop_rejected_miner { + remove_key_from_memory(&connections_key).await; + break 'outer Ok(()); + } } 50 => { // signed monitor-add event from a miner peer diff --git a/src/startup/network_broadcast.rs b/src/startup/network_broadcast.rs index 5c61681..fd63e1d 100644 --- a/src/startup/network_broadcast.rs +++ b/src/startup/network_broadcast.rs @@ -4,7 +4,7 @@ use crate::log::warn; use crate::records::memory::network_mapping::structs::{ SignedNodeEdit, NODE_ADDED_BY_OFFSET, NODE_ADDED_SIGNATURE_OFFSET, NODE_ADDED_TIMESTAMP_OFFSET, NODE_BLOCKS_MINED_OFFSET, NODE_DELETED_BLOCK_OFFSET, NODE_DELETED_TIMESTAMP_OFFSET, - NODE_IP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_RECORD_FIXED_BYTES, + NODE_IP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_PORT_OFFSET, NODE_RECORD_FIXED_BYTES, }; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::{reserve_entry, Command}; @@ -32,13 +32,17 @@ pub async fn announce_self_to_network( // explicitly request the peer's map after acceptance; incoming/server // handshakes must not import a potentially stale joining peer map. let message_type = RPC_ADD_NETWORK_NODE; - let (ip, _, _) = get_ip_and_port().await; + let (ip, port, _) = get_ip_and_port().await; + let port = port + .parse::() + .map_err(|_| "local node port was invalid".to_string())?; // Reserve a reply key so the peer's acknowledgement returns to this request. let (hashmap_key, _hashmap_tx, hashmap_rx) = reserve_entry(command_map.clone()).await; // Encode the local node identity into the same binary shape used by network commands. let ip_bytes = ip_to_binary(&ip); + let port_bytes = port.to_le_bytes(); let address_bytes = match Wallet::short_address_to_bytes(address) { Some(bytes) => bytes, None => return Err("local node address was invalid".to_string()), @@ -55,6 +59,7 @@ pub async fn announce_self_to_network( 1 + 3 + Wallet::SHORT_ADDRESS_BYTES_LENGTH + 16 + + 2 + Wallet::SHORT_ADDRESS_BYTES_LENGTH + 8 + Wallet::SIGNATURE_LENGTH, @@ -63,6 +68,7 @@ pub async fn announce_self_to_network( message.extend_from_slice(&hashmap_key); message.extend_from_slice(&address_bytes); message.extend_from_slice(&ip_bytes); + message.extend_from_slice(&port_bytes); message.extend_from_slice(&modified_by_bytes); message.extend_from_slice(&modified_timestamp_bytes); message.extend_from_slice(&modified_signature_bytes); @@ -170,7 +176,12 @@ async fn get_network_mapping_inner( let Some(address) = Wallet::bytes_to_short_address(&chunk[0..NODE_IP_OFFSET]) else { continue; }; - let ip = binary_to_ip(chunk[NODE_IP_OFFSET..NODE_BLOCKS_MINED_OFFSET].to_vec()); + let ip = binary_to_ip(chunk[NODE_IP_OFFSET..NODE_PORT_OFFSET].to_vec()); + let port = u16::from_le_bytes( + chunk[NODE_PORT_OFFSET..NODE_BLOCKS_MINED_OFFSET] + .try_into() + .unwrap(), + ); let blocks_mined = chunk[NODE_BLOCKS_MINED_OFFSET]; let added_by_bytes = &chunk[NODE_ADDED_BY_OFFSET..NODE_ADDED_TIMESTAMP_OFFSET]; let added_by = if added_by_bytes.iter().all(|&byte| byte == 0) { @@ -220,6 +231,7 @@ async fn get_network_mapping_inner( SignedNodeEdit { address: address.clone(), ip: ip.clone(), + port, modified_by: added_by, modified_timestamp: added_timestamp, modified_signature: added_signature,