contractless/Contractless
contractless
/
Contractless
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

bootstrap process changes

This commit is contained in:
viraladmin 2026-07-11 14:42:46 -06:00
parent 5c68786eef
commit 81f8f6e711
14 changed files with 193 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/records/memory/network_mapping/add.rs
View File

@ -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,13 +260,16 @@ impl NodeInfo {
if let Some(existing_node) = address_map.get(&edit.address) {
if existing_node.deleted_timestamp == 0 {
if existing_node.ip == edit.ip {
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(),
);
}
}
}
drop(address_map);
// Unsigned add requests are sponsored by the receiving node only
@ -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,

3
src/records/memory/network_mapping/mod.rs
View File

@ -36,6 +36,7 @@ static NODE_RUNTIME_STARTED_MILLIS: OnceLock<u64> = 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,

20
src/records/memory/network_mapping/persistence.rs
View File

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

15
src/records/memory/network_mapping/queries.rs
View File

@ -38,6 +38,16 @@ impl NodeInfo {
.collect()
}
pub async fn active_node_endpoints() -> Vec<String> {
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<String> {
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<Wallet>,
) -> 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
}

5
src/records/memory/network_mapping/structs.rs
View File

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

65
src/rpc/client/handshake.rs
View File

@ -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<dyn std::error::Error>> {
pub async fn connect_and_handshake(
params: Connect,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
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<dyn std::error::Error + Send + Sync>
})
}
async fn connect_and_handshake_once(
params: Connect,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// 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<dyn std::e
};
perform_handshake(handshake_params)
.await
.map_err(|err| Box::new(err) as Box<dyn std::error::Error>)
.map_err(|err| Box::new(err) as Box<dyn std::error::Error + Send + Sync>)
}
fn sponsor_endpoint_from_error(error: &str) -> Option<String> {
let (_, sponsor) = error.split_once("sponsor=")?;
let endpoint = sponsor
.split_whitespace()
.next()
.unwrap_or_default()
.trim_matches(|ch| ch == '"' || ch == '\'' || ch == ',' || ch == ';');
endpoint.parse::<SocketAddr>().ok()?;
Some(endpoint.to_string())
}
async fn connect_from_configured_ip(remote_addr: SocketAddr) -> io::Result<TcpStream> {

44
src/rpc/client/handshake_processing.rs
View File

@ -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(&current_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;
}
}

1
src/rpc/client/structs.rs
View File

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

1
src/rpc/command_maps.rs
View File

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

5
src/rpc/commands/add_network_node.rs
View File

@ -17,7 +17,7 @@ pub async fn add_network_node(
map: Arc<Mutex<Command>>,
) -> 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,

1
src/rpc/commands/mod.rs
View File

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

26
src/rpc/commands/random_node.rs
View File

@ -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<String> {
// 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::<u16>() 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 => {

23
src/rpc/server/rpc_command_loop.rs
View File

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

18
src/startup/network_broadcast.rs
View File

@ -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::<u16>()
.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,