diff --git a/src/miner/fairness.rs b/src/miner/fairness.rs index 41c9021..e0d8396 100644 --- a/src/miner/fairness.rs +++ b/src/miner/fairness.rs @@ -1,3 +1,4 @@ +use crate::common::types::BLOCKS_PER_HALVING; use crate::log::info; use crate::miner::flag::{is_mining_stop_requested, is_normal_mode, set_mining_state, MiningState}; use crate::records::block_height::get_block_height::get_height; @@ -8,12 +9,12 @@ use crate::sleep; use crate::Duration; pub async fn fairness_difficulty(block_height: u32, miner_wallet: &str) -> bool { - // The fairness window tightens once peri tenth of the first halving interval. - let current_interval = (block_height / 960000) + 1; + // The fairness window tightens once per completed halving interval. + let current_interval = block_height / BLOCKS_PER_HALVING; // Start by allowing ten consecutive blocks, then reduce the // allowance over time without ever dropping below one. - let max_consecutive_blocks = u8::max(10 - current_interval as u8, 1) as u32; + let max_consecutive_blocks = 10_u32.saturating_sub(current_interval).max(1); // Only the trailing fairness window needs to be checked. let start_block = block_height.saturating_sub(max_consecutive_blocks); diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index fee450c..e77fb97 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -166,7 +166,7 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { wallet: context.wallet.clone(), db: context.db.clone(), map: context.map.clone(), - first: false, + first: peer_connection_count().await == 0, }; match connect_and_handshake(connect).await { @@ -312,16 +312,19 @@ impl Connection { }); } if let Some(connection_info) = removed.as_ref() { - let client_role = ClientType::from_bytes(&connection_info.client_type) - .map(|client_type| client_type.as_str()) - .unwrap_or("unknown"); - info!( - "[connection_manager] connection dropped: role={} direction={} peer={}:{}", - client_role, - connection_type.as_str(), - ip, - port - ); + let client_type = ClientType::from_bytes(&connection_info.client_type); + if client_type != Some(ClientType::Client) { + let client_role = client_type + .map(|client_type| client_type.as_str()) + .unwrap_or("unknown"); + info!( + "[connection_manager] connection dropped: role={} direction={} peer={}:{}", + client_role, + connection_type.as_str(), + ip, + port + ); + } } removed } @@ -927,6 +930,37 @@ pub async fn peer_is_operational(key: &str) -> bool { .unwrap_or(false) } +pub async fn peer_accepts_live_relay(key: &str) -> bool { + let Some((ip, port)) = split_ip_port_key(key) else { + return false; + }; + let ip_bytes = ip_to_binary(&ip); + + CONNECTIONS + .read() + .await + .as_ref() + .and_then(|connection| { + connection + .connection_map + .iter() + .find_map(|(connection_key, info)| { + if connection_key.ip == ip_bytes + && connection_key.port == port + && ClientType::from_bytes(&info.client_type) == Some(ClientType::Miner) + { + Some( + info.ready + || (info.wallet_registry_synced && info.network_map_synced), + ) + } else { + None + } + }) + }) + .unwrap_or(false) +} + pub async fn live_miner_peer_streams() -> Vec<(String, Arc>)> { // Snapshot consensus and recovery checks vote only across currently // connected miner peers, regardless of incoming/outgoing direction. diff --git a/src/rpc/client/genesis_compat.rs b/src/rpc/client/genesis_compat.rs index 659b4f4..38d20c1 100644 --- a/src/rpc/client/genesis_compat.rs +++ b/src/rpc/client/genesis_compat.rs @@ -28,7 +28,11 @@ pub async fn ensure_compatible_genesis( match request_block_hash_at_height(stream, map, connections_key.clone(), 0).await { Ok(hash) => Some(hash), Err(err) if err.contains("Block 0 not found") => None, - Err(err) if remote_height == 0 && err.contains("Node is syncing") => None, + Err(err) if err.contains("Node is syncing") => { + return Err(format!( + "remote peer {connections_key} is syncing; cannot verify genesis before sync" + )); + } Err(err) => { return Err(format!( "remote genesis compatibility check failed for {connections_key}: {err}" diff --git a/src/rpc/client/syncing.rs b/src/rpc/client/syncing.rs index 7667470..fd54ebd 100644 --- a/src/rpc/client/syncing.rs +++ b/src/rpc/client/syncing.rs @@ -7,6 +7,7 @@ use crate::records::block_height::get_block_height::get_height; use crate::records::memory::response_channels::reserve_entry_with_context; use crate::records::memory::response_channels::Command; use crate::records::memory::torrent_status::{set_torrent_status, TorrentStatus}; +use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::command_maps::RPC_TORRENT_BY_HEIGHT; use crate::sled::Db; use crate::timeout; @@ -159,6 +160,15 @@ pub async fn node_syncing( wallet: Arc, connections_key: String, ) -> io::Result<()> { + ensure_compatible_genesis( + stream.clone(), + map.clone(), + connections_key.clone(), + remote_height, + ) + .await + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?; + let mut local_height = get_height(db); // Normalize the local height into the next expected block height // so sync requests start at the first missing block. diff --git a/src/rpc/commands/block_hash_at_height.rs b/src/rpc/commands/block_hash_at_height.rs index 2797749..bd762f6 100644 --- a/src/rpc/commands/block_hash_at_height.rs +++ b/src/rpc/commands/block_hash_at_height.rs @@ -4,7 +4,7 @@ use crate::records::unpack_block::unpack_header::load_block_header; use crate::rpc::responses::RpcResponse; pub async fn request_block_hash_at_height(block_number: u32) -> RpcResponse { - if is_syncing_mode() { + if is_syncing_mode() && block_number != 0 { return RpcResponse::Binary(b"error: Node is syncing".to_vec()); } diff --git a/src/rpc/server/command_loop_state.rs b/src/rpc/server/command_loop_state.rs index 622d14c..206badf 100644 --- a/src/rpc/server/command_loop_state.rs +++ b/src/rpc/server/command_loop_state.rs @@ -1,9 +1,10 @@ use crate::io::ErrorKind; use crate::log::warn; -use crate::records::memory::connections::get_client_type_from_memory; +use crate::records::memory::connections::{get_client_type_from_memory, peer_accepts_live_relay}; use crate::records::memory::enums::ClientType; use crate::rpc::command_maps::{ - RPC_BLOCK_HEIGHT, RPC_BLOCK_PIECE, RPC_REPLY, RPC_TORRENT_BY_HEIGHT, + RPC_BLOCK_HEIGHT, RPC_BLOCK_PIECE, RPC_NETWORK_MONITOR_ADD, RPC_NETWORK_MONITOR_REMOVE, + RPC_REPLY, RPC_SUBMIT_TORRENT, RPC_SUBMIT_TRANSACTION, RPC_TORRENT_BY_HEIGHT, }; use crate::rpc::server::connection_memory_manager::remove_stream_from_memory; use crate::rpc::server::flood_protection::check_request_frequency_with_client_type; @@ -72,6 +73,16 @@ fn skip_flood_check(command: u8, client_type: ClientType) -> bool { ) } +fn requires_operational_miner(command: u8) -> bool { + matches!( + command, + RPC_SUBMIT_TRANSACTION + | RPC_SUBMIT_TORRENT + | RPC_NETWORK_MONITOR_ADD + | RPC_NETWORK_MONITOR_REMOVE + ) +} + pub async fn next_incoming_command( stream_locked: Arc>, db: &Db, @@ -91,6 +102,17 @@ pub async fn next_incoming_command( .await .unwrap_or(ClientType::Miner); + if client_type == ClientType::Miner + && requires_operational_miner(command) + && !peer_accepts_live_relay(connections_key).await + { + warn!( + "[rpc] closing unsynced miner peer that sent relay command: peer={connections_key} cmd={command}" + ); + remove_stream_from_memory(&stream_locked).await; + return Ok(None); + } + // Replies and miner sync traffic belong to expected node-to-node // maintenance paths. All other commands still count against flood // protection, including non-sync commands from miners. diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index c5561cb..42cc1c3 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -77,9 +77,11 @@ async fn sync_incoming_peer_before_operational( if local_genesis_exists && remote_height < local_height { warn!( - "[startup] incoming peer is behind local chain and will remain non-operational until it catches up: local_height={local_height} remote_height={remote_height}" + "[startup] incoming peer is behind local chain and must reconnect through its own sync path: local_height={local_height} remote_height={remote_height}" ); - return Ok(false); + return Err(format!( + "incoming peer is behind local chain: local_height={local_height} remote_height={remote_height}" + )); } if !local_genesis_exists || remote_height > local_height + 10 { diff --git a/src/startup/connections.rs b/src/startup/connections.rs index b9e06c7..e087778 100644 --- a/src/startup/connections.rs +++ b/src/startup/connections.rs @@ -4,7 +4,7 @@ use crate::miner::flag::{ clear_mining_stop_request, is_normal_mode, set_mining_state, set_node_mode, MiningState, NodeMode, }; -use crate::records::memory::connections::{miner_connection_count, peer_connection_count}; +use crate::records::memory::connections::peer_connection_count; use crate::records::memory::response_channels::Command; use crate::rpc::client::handshake::connect_and_handshake; use crate::rpc::client::structs::Connect; @@ -137,10 +137,6 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc, map: Arc 0 { - continue; - } - info!("[reconnect] no operational peers remain; retrying bootstrap recovery"); match attempt_bootstrap_connections( db.clone(),