From 291fbd5755dd8f8e4636e0e538ccca41b26f8bb8 Mon Sep 17 00:00:00 2001 From: contractless <00.pur.ple@gmail.com> Date: Fri, 11 Sep 2026 17:22:31 -0600 Subject: [PATCH] recovery fixes --- src/miner/flag.rs | 27 ++ src/records/memory/connections.rs | 105 ++++++- src/records/memory/network_mapping/add.rs | 112 ++++--- src/records/memory/network_mapping/mod.rs | 15 - src/records/memory/network_mapping/queries.rs | 6 + src/records/memory/response_channels.rs | 13 +- src/records/mod.rs | 1 + src/records/peer_endpoints.rs | 130 +++++++++ src/rpc/client/handshake_processing.rs | 275 ++++++++++++++---- src/rpc/server/connection_memory_manager.rs | 73 ++++- src/rpc/server/flood_protection.rs | 102 ++++++- src/rpc/server/handshake.rs | 92 +++++- src/startup/connections.rs | 81 ++++-- src/startup/initialize_startup.rs | 2 - src/startup/network_broadcast.rs | 8 + .../torrenting_system/download_pieces.rs | 22 +- src/torrent/torrenting_system/get_nodes.rs | 13 + 17 files changed, 913 insertions(+), 164 deletions(-) create mode 100644 src/records/peer_endpoints.rs diff --git a/src/miner/flag.rs b/src/miner/flag.rs index b2478d9..0f7b3ec 100644 --- a/src/miner/flag.rs +++ b/src/miner/flag.rs @@ -159,6 +159,33 @@ pub async fn begin_chain_sync() -> ChainOperationGuard { ChainOperationGuard { active: true } } +// Incoming setup belongs to one socket, not to an endpoint that may have +// reconnected. Check its identity while holding the connection read lock so +// removal cannot race with admission to the chain gate and the stop request. +pub async fn begin_connected_peer_sync( + stream: &crate::Arc>, +) -> Option { + loop { + let connections = crate::records::memory::connections::CONNECTIONS.read().await; + let current = connections.as_ref().is_some_and(|connection| { + connection.connection_map.values().any(|info| crate::Arc::ptr_eq(&info.stream, stream)) + }); + if !current { + return None; + } + if try_acquire_chain_operation_gate() { + let guard = ChainOperationGuard { active: true }; + request_mining_stop(); + drop(connections); + wait_for_mining_idle().await; + set_node_mode(NodeMode::Syncing); + return Some(guard); + } + drop(connections); + sleep(Duration::from_millis(5)).await; + } +} + pub async fn begin_reorg_lock() -> ChainOperationGuard { // Orphan correction shares the canonical chain gate with synchronization, // so rollback/replay cannot overlap a startup or live catch-up save. diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 7e822a5..36917d9 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -6,11 +6,13 @@ use crate::records::memory::network_mapping::monitor::{MONITOR_ACTION_ADD, MONIT use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit}; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::{ - delete_entry, reserve_entry_with_context, Command, + delete_entry, reserve_entry_with_context, retire_peer_entries_created_before, Command, }; use crate::records::memory::structs::{Connection, ConnectionHealth, StoreConnectionParams}; use crate::rpc::client::handshake::connect_and_handshake; -use crate::rpc::client::handshake_processing::{bootstrap_peer_discovery, BootstrapParams}; +use crate::rpc::client::handshake_processing::{ + bootstrap_peer_discovery, startup_discovery_active, BootstrapParams, +}; use crate::rpc::client::structs::Connect; use crate::rpc::command_maps::{RPC_BLOCK_HEIGHT, RPC_SETUP_COMPLETE}; use crate::rpc::responses::RpcResponse; @@ -124,6 +126,17 @@ fn node_runtime_context() -> Option { .and_then(|context| context.clone()) } +pub(crate) fn preferred_miner_connection_type( + peer_wallet_short_address: &str, +) -> Option { + let local_wallet = node_runtime_context()?.wallet.saved.short_address.clone(); + if local_wallet.as_str() < peer_wallet_short_address { + Some(ConnectionType::Outgoing) + } else { + Some(ConnectionType::Incoming) + } +} + pub(crate) fn spawn_outage_recovery_from_runtime() { let Some(context) = node_runtime_context() else { return; @@ -185,6 +198,7 @@ async fn reconnect_replacement_inner(excluded_ip: &str) { } } +#[allow(dead_code)] // Retained while direct dropped-peer retries are disabled below. async fn retry_dropped_outgoing(ip: String, port: u16) { let recovery_key = format!("peer:{ip}"); let Some(_recovery_guard) = try_start_recovery(recovery_key) else { @@ -199,7 +213,7 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { // Losing every miner peer is process-level outage recovery, not an // independent retry of each socket that happened to time out. - if miner_connection_count().await == 0 { + if sync_capable_miner_connection_count().await == 0 { spawn_outage_recovery(context.db, context.wallet, context.map); return; } @@ -236,10 +250,7 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { wallet: context.wallet.clone(), db: context.db.clone(), map: context.map.clone(), - // A miner stream that is still completing setup already has a - // startup owner. Counting only ready peers could launch a second - // full startup synchronization while the first one is active. - first: miner_connection_count().await == 0, + first: sync_capable_miner_connection_count().await == 0 && !startup_discovery_active(), }; match connect_and_handshake(connect).await { @@ -268,9 +279,13 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { } pub fn spawn_retry_dropped_outgoing(ip: String, port: u16) { - tokio::spawn(async move { - retry_dropped_outgoing(ip, port).await; - }); + // Temporarily disabled: let the disconnected node initiate recovery. + // Keep this common entry point so all drop paths (including penalties) + // follow the same policy, and restoring retries only requires uncommenting. + let _ = (ip, port); + // tokio::spawn(async move { + // retry_dropped_outgoing(ip, port).await; + // }); } pub fn spawn_reconnect_bootstrap(params: BootstrapParams) { @@ -387,7 +402,13 @@ impl Connection { ); } let stream = Arc::clone(&connection_info.stream); + let peer = format!("{ip}:{port}"); + let command_map = node_runtime_context().map(|context| context.map); + let retirement_cutoff = crate::Instant::now(); tokio::spawn(async move { + if let Some(command_map) = command_map { + retire_peer_entries_created_before(command_map, &peer, retirement_cutoff).await; + } let mut stream_guard = stream.lock().await; let _ = stream_guard.shutdown().await; }); @@ -528,7 +549,7 @@ impl Connection { if connection_type == ConnectionType::Outgoing { spawn_retry_dropped_outgoing(ip.clone(), port); } - if miner_connection_count().await == 0 { + if sync_capable_miner_connection_count().await == 0 { spawn_outage_recovery_from_runtime(); } break; @@ -757,6 +778,20 @@ impl Connection { .count() } + pub fn count_sync_capable_miner_connections(&self) -> usize { + self.connection_map + .values() + .filter(|info| { + ClientType::from_bytes(&info.client_type) == Some(ClientType::Miner) + && (info.ready || (info.wallet_registry_synced && info.network_map_synced)) + && matches!( + info.health, + ConnectionHealth::Connected | ConnectionHealth::Busy + ) + }) + .count() + } + // Return ready peer streams so broadcast-style paths do not send // network-wide traffic to peers still completing startup sync. pub fn get_all_streams(&self) -> Vec>> { @@ -835,6 +870,30 @@ impl Connection { .collect() } + pub fn get_sync_capable_peer_streams_with_keys(&self) -> Vec<(String, Arc>)> { + self.connection_map + .iter() + .filter_map(|(key, connection_info)| { + if ClientType::from_bytes(&connection_info.client_type) != Some(ClientType::Miner) + || (!connection_info.ready + && !(connection_info.wallet_registry_synced + && connection_info.network_map_synced)) + || !matches!( + connection_info.health, + ConnectionHealth::Connected | ConnectionHealth::Busy + ) + { + return None; + } + let ip = binary_to_ip(key.ip.clone()); + Some(( + format!("{}:{}", ip, key.port), + Arc::clone(&connection_info.stream), + )) + }) + .collect() + } + // Resolve a stored outgoing node connection back to its live stream. pub fn get_stream_for_outgoing(&self, ip: &str, port: u16) -> Option>> { let ip_bytes = ip_to_binary(ip); @@ -1114,6 +1173,12 @@ fn spawn_monitor_activation( ConnectionType::from_bytes(&key.connection_type).unwrap_or(ConnectionType::Incoming) }; + if let Err(err) = crate::records::peer_endpoints::remember_peer_endpoints( + &context.db, + &[format!("{ip}:{port}")], + ) { + warn!("[reconnect] could not cache operational peer endpoint: {err}"); + } Connection::client_checkup(stream, connection_type, ip, port, command_map); }); } @@ -1268,6 +1333,15 @@ pub async fn miner_connection_count() -> usize { .unwrap_or(0) } +pub async fn sync_capable_miner_connection_count() -> usize { + CONNECTIONS + .read() + .await + .as_ref() + .map(|connection| connection.count_sync_capable_miner_connections()) + .unwrap_or(0) +} + pub async fn mark_peer_wallet_registry_synced(key: &str, stream: &Arc>) -> bool { CONNECTIONS .write() @@ -1616,6 +1690,15 @@ pub async fn startup_synced_peer_streams() -> Vec<(String, Arc> .unwrap_or_default() } +pub async fn sync_capable_peer_streams() -> Vec<(String, Arc>)> { + CONNECTIONS + .read() + .await + .as_ref() + .map(|connection| connection.get_sync_capable_peer_streams_with_keys()) + .unwrap_or_default() +} + pub async fn get_client_type_from_memory(key: &str) -> Option { // Recover the stored client role from the serialized connection key // used throughout the RPC layer. diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index af3c1af..8f81838 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -21,10 +21,27 @@ fn signature_is_empty(signature: &str) -> bool { } impl NodeInfo { + fn new_node_admission_limit_reached( + address_map: &HashMap, + edit: &SignedNodeEdit, + current_timestamp: u64, + ) -> bool { + // Reconnecting a known wallet does not grow the membership map. All + // signature, sponsor, deletion-order and IP-ownership checks still apply. + !address_map.contains_key(&edit.address) + && address_map + .values() + .filter(|node| { + node.added_by == edit.modified_by + && current_timestamp.saturating_sub(node.added_timestamp) <= ONE_HOUR_MILLIS + }) + .count() + >= 10 + } + fn eligible_sponsor_endpoint_from_map( address_map: &HashMap, operational_addresses: &HashSet, - current_timestamp: u64, excluded_address: &str, ) -> Option { let mut candidates: Vec<(&String, &NodeInfo)> = address_map @@ -33,7 +50,6 @@ impl NodeInfo { address.as_str() != excluded_address && operational_addresses.contains(address.as_str()) && node.deleted_timestamp == 0 - && current_timestamp.saturating_sub(node.added_timestamp) >= ONE_HOUR_MILLIS && node.blocks_mined >= 100 }) .collect(); @@ -50,13 +66,11 @@ impl NodeInfo { fn sponsor_redirect_suffix( address_map: &HashMap, operational_addresses: &HashSet, - current_timestamp: u64, excluded_address: &str, ) -> String { Self::eligible_sponsor_endpoint_from_map( address_map, operational_addresses, - current_timestamp, excluded_address, ) .map(|endpoint| format!(" sponsor={endpoint}")) @@ -921,8 +935,8 @@ impl NodeInfo { { let mut address_map = ADDRESS_MAP.lock().await; - // Once the chain is mature, adding nodes is restricted to older - // active participants with sufficient mined history. + // Once the chain is mature, adding nodes is restricted to active + // participants with sufficient mined history. let local_height = get_height(&db); if !Self::self_add_allowed_at_height(local_height) { let signer_key = Wallet::normalize_to_short_address(&edit.modified_by) @@ -930,30 +944,21 @@ impl NodeInfo { let signer_node = address_map.get(&signer_key); let signer_is_local = signer_key == wallet.saved.short_address; let valid_added_by = signer_node - .map(|node| { - current_timestamp.saturating_sub(node.added_timestamp) >= ONE_HOUR_MILLIS - && node.deleted_timestamp == 0 - && (!signer_is_local - || Self::local_runtime_is_mature(current_timestamp)) - }) + .map(|node| node.deleted_timestamp == 0) .unwrap_or(false); if !valid_added_by { let signer_exists = signer_node.is_some(); let signer_deleted_timestamp = signer_node.map(|node| node.deleted_timestamp).unwrap_or(0); - let signer_age_ms = signer_node - .map(|node| current_timestamp.saturating_sub(node.added_timestamp)) - .unwrap_or(0); let signer_blocks_mined = signer_node.map(|node| node.blocks_mined).unwrap_or(0); let redirect = Self::sponsor_redirect_suffix( &address_map, &operational_addresses, - current_timestamp, &signer_key, ); return RpcResponse::Binary(format!( - "Error: This address cannot add nodes. It must exist for at least 60 minutes and not be marked for deletion local_height={local_height} self_add_block={} self_add_limit={} signer={signer_key} signer_exists={signer_exists} signer_is_local={signer_is_local} signer_deleted_timestamp={signer_deleted_timestamp} signer_age_ms={signer_age_ms} signer_blocks_mined={signer_blocks_mined}{redirect}", + "Error: This address cannot add nodes because it is missing or marked for deletion local_height={local_height} self_add_block={} self_add_limit={} signer={signer_key} signer_exists={signer_exists} signer_is_local={signer_is_local} signer_deleted_timestamp={signer_deleted_timestamp} signer_blocks_mined={signer_blocks_mined}{redirect}", super::SELF_ADD_BLOCK, Self::self_add_limit_height(), ) @@ -964,7 +969,6 @@ impl NodeInfo { let redirect = Self::sponsor_redirect_suffix( &address_map, &operational_addresses, - current_timestamp, &signer_key, ); return RpcResponse::Binary(format!( @@ -973,15 +977,7 @@ impl NodeInfo { } } - let added_by_count_in_last_hour = address_map - .values() - .filter(|node| { - node.added_by == edit.modified_by - && current_timestamp.saturating_sub(node.added_timestamp) <= ONE_HOUR_MILLIS - }) - .count(); - - if added_by_count_in_last_hour >= 10 { + if Self::new_node_admission_limit_reached(&address_map, &edit, current_timestamp) { return RpcResponse::Binary( b"Error: Cannot add more than 10 nodes in 60 minutes".to_vec(), ); @@ -1144,6 +1140,53 @@ impl NodeInfo { mod tests { use super::*; + #[test] + fn full_admission_window_allows_known_wallet_but_not_a_new_wallet_on_same_ip() { + let mut map = HashMap::new(); + for index in 0..10 { + map.insert( + format!("node-{index}"), + NodeInfo::new( + format!("8.8.8.{}", index + 1), + 50050, + 0, + "sponsor".into(), + 100, + "signature".into(), + ), + ); + } + let mut edit = SignedNodeEdit { + address: "node-0".into(), + ip: "8.8.8.1".into(), + port: 50050, + modified_by: "sponsor".into(), + modified_timestamp: 200, + modified_signature: "signature".into(), + }; + map.get_mut("node-0").unwrap().deleted_timestamp = 150; + map.get_mut("node-0").unwrap().deleted_block = 50; + assert!(!NodeInfo::new_node_admission_limit_reached( + &map, &edit, 200 + )); + edit.address = "replacement-wallet".into(); + assert!(NodeInfo::new_node_admission_limit_reached(&map, &edit, 200)); + assert!(NodeInfo::new_node_admission_limit_reached( + &map, + &edit, + 100 + ONE_HOUR_MILLIS + )); + assert!(!NodeInfo::new_node_admission_limit_reached( + &map, + &edit, + 101 + ONE_HOUR_MILLIS + )); + map.remove("node-9"); + assert!(!NodeInfo::new_node_admission_limit_reached( + &map, &edit, 200 + )); + } + fn snapshot_node(ip: &str, timestamp: u64, monitors: &[&str]) -> NodeInfo { let mut node = NodeInfo::new( ip.to_string(), @@ -1181,7 +1224,6 @@ mod tests { #[test] fn sponsor_redirect_selects_an_eligible_different_node() { - let now = ONE_HOUR_MILLIS * 3; let mut map = HashMap::new(); let mut rejected = snapshot_node("1.2.3.4", 1, &[]); @@ -1202,7 +1244,6 @@ mod tests { NodeInfo::eligible_sponsor_endpoint_from_map( &map, &HashSet::from(["newer.cltc".to_string(), "older.cltc".to_string()]), - now, "rejected.cltc", ), Some("1.2.3.5:50051".to_string()) @@ -1210,7 +1251,7 @@ mod tests { } #[test] - fn sponsor_redirect_ignores_deleted_young_and_under_mined_nodes() { + fn sponsor_redirect_allows_recent_active_mined_node() { let now = ONE_HOUR_MILLIS * 3; let mut map = HashMap::new(); @@ -1219,9 +1260,9 @@ mod tests { deleted.deleted_timestamp = now; map.insert("deleted.cltc".to_string(), deleted); - let mut young = snapshot_node("1.2.3.5", now - 1_000, &[]); - young.blocks_mined = 100; - map.insert("young.cltc".to_string(), young); + let mut recent = snapshot_node("1.2.3.5", now - 1_000, &[]); + recent.blocks_mined = 100; + map.insert("recent.cltc".to_string(), recent); let mut under_mined = snapshot_node("1.2.3.6", 1, &[]); under_mined.blocks_mined = 99; @@ -1232,13 +1273,12 @@ mod tests { &map, &HashSet::from([ "deleted.cltc".to_string(), - "young.cltc".to_string(), + "recent.cltc".to_string(), "under-mined.cltc".to_string(), ]), - now, "none.cltc", ), - None + Some("1.2.3.5:50050".to_string()) ); } diff --git a/src/records/memory/network_mapping/mod.rs b/src/records/memory/network_mapping/mod.rs index 62d1028..329218e 100644 --- a/src/records/memory/network_mapping/mod.rs +++ b/src/records/memory/network_mapping/mod.rs @@ -19,7 +19,6 @@ use crate::wallets::structures::Wallet; use crate::Arc; use crate::HashMap; use crate::Mutex; -use crate::OnceLock; use crate::Utc; use std::collections::VecDeque; @@ -47,8 +46,6 @@ lazy_static! { pub const SELF_ADD_BLOCK: u32 = 0; pub const SELF_ADD_WINDOW_BLOCKS: u32 = 10_000; -static NODE_RUNTIME_STARTED_MILLIS: OnceLock = OnceLock::new(); - #[derive(Debug)] pub struct NodeInfo { ip: String, @@ -63,18 +60,6 @@ pub struct NodeInfo { } impl NodeInfo { - pub fn initialize_runtime_start() { - let _ = NODE_RUNTIME_STARTED_MILLIS.set(Utc::now().timestamp_millis() as u64); - } - - fn runtime_started_millis() -> u64 { - *NODE_RUNTIME_STARTED_MILLIS.get_or_init(|| Utc::now().timestamp_millis() as u64) - } - - fn local_runtime_is_mature(current_timestamp: u64) -> bool { - current_timestamp.saturating_sub(Self::runtime_started_millis()) >= 3_600_000 - } - pub fn self_add_allowed_at_height(height: u32) -> bool { height <= SELF_ADD_BLOCK.saturating_add(SELF_ADD_WINDOW_BLOCKS) } diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index f21da3f..22153c7 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -13,6 +13,12 @@ use crate::sled::Db; use std::collections::HashSet; impl NodeInfo { + pub(crate) async fn mapping_sync_record_count() -> usize { + let memberships = ADDRESS_MAP.lock().await.len(); + let monitor_events = MONITOR_EVENT_STATE.lock().await.len(); + memberships.saturating_add(monitor_events) + } + fn eligible_at_block(node_info: &NodeInfo, block_number: u32) -> bool { node_info.deleted_timestamp == 0 || (node_info.deleted_block > 0 && block_number < node_info.deleted_block) diff --git a/src/records/memory/response_channels.rs b/src/records/memory/response_channels.rs index 27c1be2..d96002a 100644 --- a/src/records/memory/response_channels.rs +++ b/src/records/memory/response_channels.rs @@ -198,11 +198,22 @@ pub async fn delete_entry(map: Arc>, key: Byte3) { } pub async fn retire_peer_entries(map: Arc>, peer: &str) { + retire_peer_entries_created_before(map, peer, Instant::now()).await; +} + +pub async fn retire_peer_entries_created_before( + map: Arc>, + peer: &str, + cutoff: Instant, +) { let mut map = map.lock().await; let expires_at = Instant::now() + Duration::from_secs(30); for channel_pair in map.values_mut() { - if channel_pair.peer.as_deref() != Some(peer) || channel_pair.expires_at.is_some() { + if channel_pair.peer.as_deref() != Some(peer) + || channel_pair.expires_at.is_some() + || channel_pair.created_at > cutoff + { continue; } diff --git a/src/records/mod.rs b/src/records/mod.rs index 0c078c6..cfb941a 100644 --- a/src/records/mod.rs +++ b/src/records/mod.rs @@ -3,6 +3,7 @@ pub mod balance_sheet; pub mod block_height; pub mod ip_score; pub mod memory; +pub mod peer_endpoints; pub mod record_chain; pub mod unpack_block; pub mod wallet_registry; diff --git a/src/records/peer_endpoints.rs b/src/records/peer_endpoints.rs new file mode 100644 index 0000000..414a37b --- /dev/null +++ b/src/records/peer_endpoints.rs @@ -0,0 +1,130 @@ +//! Durable dial hints only. These records never establish membership or mining eligibility. +use crate::common::network_startup::canonical_miner_endpoint; +use crate::sled::Db; + +const TREE: &str = "peer_endpoint_cache"; +const KEY: &[u8] = b"endpoints_v1"; +const MAX_ENDPOINTS: usize = 512; + +fn merge_endpoints(preferred: &[String], previous: &[String]) -> Vec { + let mut endpoints = Vec::new(); + for value in preferred.iter().chain(previous) { + if let Some(endpoint) = canonical_miner_endpoint(value) { + if !endpoints.contains(&endpoint) { + endpoints.push(endpoint); + } + if endpoints.len() == MAX_ENDPOINTS { + break; + } + } + } + endpoints +} + +pub fn remember_peer_endpoints(db: &Db, endpoints: &[String]) -> sled::Result<()> { + let preferred = merge_endpoints(endpoints, &[]); + if preferred.is_empty() { + return Ok(()); + } + // Atomic read/modify/write preserves discoveries from concurrent peer setups. + db.open_tree(TREE)?.update_and_fetch(KEY, |previous| { + let previous: Vec = previous + .and_then(|bytes| serde_json::from_slice(bytes).ok()) + .unwrap_or_default(); + Some(serde_json::to_vec(&merge_endpoints(&preferred, &previous)).unwrap()) + })?; + Ok(()) +} + +pub fn remembered_peer_endpoints(db: &Db) -> sled::Result> { + let previous: Vec = db + .open_tree(TREE)? + .get(KEY)? + .and_then(|bytes| serde_json::from_slice(&bytes).ok()) + .unwrap_or_default(); + // Revalidate disk hints and tolerate malformed cache data independently of chain state. + Ok(merge_endpoints(&previous, &[])) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hints_survive_reopening_without_restoring_membership() { + let path = + std::env::temp_dir().join(format!("contractless-peers-{}", rand::random::())); + { + let db = sled::open(&path).unwrap(); + remember_peer_endpoints(&db, &["8.8.8.8:50050".into()]).unwrap(); + db.flush().unwrap(); + } + { + let db = sled::open(&path).unwrap(); + assert_eq!( + remembered_peer_endpoints(&db).unwrap(), + vec!["8.8.8.8:50050"] + ); + assert!(db.get(b"height").unwrap().is_none()); + } + std::fs::remove_dir_all(path).unwrap(); + } + + #[test] + fn cache_is_bounded_deduplicated_and_filters_invalid_endpoints() { + let db = sled::Config::new().temporary(true).open().unwrap(); + let endpoints = (1..=600) + .map(|port| format!("8.8.8.8:{port}")) + .collect::>(); + remember_peer_endpoints(&db, &endpoints).unwrap(); + remember_peer_endpoints( + &db, + &[ + "1.1.1.1:50050".into(), + "1.1.1.1:50050".into(), + "127.0.0.1:50050".into(), + "10.0.0.1:50050".into(), + "8.8.4.4:0".into(), + "bad".into(), + ], + ) + .unwrap(); + let peers = remembered_peer_endpoints(&db).unwrap(); + assert_eq!(peers.len(), MAX_ENDPOINTS); + assert_eq!(peers[0], "1.1.1.1:50050"); + assert_eq!(peers[1], "8.8.8.8:1"); + assert!(!peers + .iter() + .any(|p| p.starts_with("127.") || p.starts_with("10."))); + } + + #[test] + fn concurrent_discoveries_do_not_overwrite_each_other() { + let db = sled::Config::new().temporary(true).open().unwrap(); + let workers = (1..=16) + .map(|port| { + let db = db.clone(); + std::thread::spawn(move || { + remember_peer_endpoints(&db, &[format!("8.8.8.8:{port}")]).unwrap(); + }) + }) + .collect::>(); + for worker in workers { + worker.join().unwrap(); + } + let peers = remembered_peer_endpoints(&db).unwrap(); + assert_eq!(peers.len(), 16); + for port in 1..=16 { + assert!(peers.contains(&format!("8.8.8.8:{port}"))); + } + } + + #[test] + fn corrupt_cache_does_not_block_startup_or_future_discovery() { + let db = sled::Config::new().temporary(true).open().unwrap(); + db.open_tree(TREE).unwrap().insert(KEY, b"broken").unwrap(); + assert!(remembered_peer_endpoints(&db).unwrap().is_empty()); + remember_peer_endpoints(&db, &["8.8.8.8:50050".into()]).unwrap(); + assert_eq!(remembered_peer_endpoints(&db).unwrap().len(), 1); + } +} diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 2bdd31c..45a2e1e 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -12,7 +12,8 @@ use crate::orphans::torrent_candidates::hydrate_torrent_candidates; use crate::records::block_height::get_block_height::get_height; use crate::records::memory::connections::{ mark_peer_network_map_synced, mark_peer_operational, mark_peer_wallet_registry_synced, - spawn_peer_setup_retry, startup_synced_peer_streams, CONNECTIONS, + spawn_outage_recovery_from_runtime, spawn_peer_setup_retry, startup_synced_peer_streams, + sync_capable_peer_streams, CONNECTIONS, }; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; @@ -50,6 +51,10 @@ use tokio::time::Instant as TokioInstant; static STARTUP_DISCOVERY_ACTIVE: AtomicBool = AtomicBool::new(false); +pub fn startup_discovery_active() -> bool { + STARTUP_DISCOVERY_ACTIVE.load(Ordering::SeqCst) +} + struct StartupDiscoveryGuard; impl Drop for StartupDiscoveryGuard { @@ -107,6 +112,38 @@ pub fn spawn_bootstrap_peer_discovery(params: BootstrapParams) { }); } +async fn wait_for_replacement_sync_peer( + failed_stream: &Arc>, +) -> (String, Arc>) { + let mut recovery_ticks = 0_u16; + loop { + if let Some(peer) = sync_capable_peer_streams() + .await + .into_iter() + .find(|(_, stream)| !Arc::ptr_eq(stream, failed_stream)) + { + return peer; + } + if recovery_ticks % 20 == 0 { + spawn_outage_recovery_from_runtime(); + } + recovery_ticks = recovery_ticks.wrapping_add(1); + sleep(Duration::from_millis(250)).await; + } +} + +fn is_peer_transport_failure(err: &str) -> bool { + err.contains("Timed out") + || err.contains("timed out") + || err.contains("Connection reset") + || err.contains("Connection aborted") + || err.contains("Broken pipe") + || err.contains("connection closed") + || err.contains("No response received") + || err.contains("response channel closed") + || err.contains("Failed to send") +} + pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), String> { let chain_sync_guard = if params.run_startup_sync { Some(begin_chain_sync().await) @@ -115,8 +152,8 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str }; let (_, _, local_endpoint) = get_ip_and_port().await; let max = SETTINGS.outgoing_connections; - let current_key = params.connections_key.clone(); - let stream = params.stream; + let mut current_key = params.connections_key.clone(); + let mut stream = params.stream; let mut imported_chain = false; if params.run_startup_sync { @@ -179,13 +216,13 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str } if !setup_tasks.is_empty() { - let warmup_deadline = TokioInstant::now() + Duration::from_secs(10); + let warmup_deadline = TokioInstant::now() + Duration::from_secs(30); while TokioInstant::now() < warmup_deadline && setup_tasks.iter().any(|task| !task.is_finished()) { sleep(Duration::from_millis(100)).await; } - let pool_size = startup_synced_peer_streams().await.len(); + let pool_size = sync_capable_peer_streams().await.len(); info!( "[sync] torrent peer pool warmed with {pool_size} synchronized peer(s); starting chain synchronization" ); @@ -198,22 +235,50 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str if params.run_startup_sync { loop { let local_height = get_height(¶ms.db); - let remote_height = - request_remote_height(stream.clone(), params.map.clone(), current_key.clone()) - .await?; - ensure_compatible_genesis( + let remote_height = match request_remote_height( + stream.clone(), + params.map.clone(), + current_key.clone(), + ) + .await + { + Ok(height) => height, + Err(err) => { + warn!( + "[sync] source peer {current_key} failed height request; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = wait_for_replacement_sync_peer(&failed_stream).await; + info!("[sync] continuing startup synchronization through {current_key}"); + continue; + } + }; + if let Err(err) = ensure_compatible_genesis( stream.clone(), params.map.clone(), current_key.clone(), remote_height, ) - .await?; + .await + { + if !is_peer_transport_failure(&err) { + return Err(err); + } + warn!( + "[sync] source peer {current_key} failed genesis verification transport; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = wait_for_replacement_sync_peer(&failed_stream).await; + continue; + } let local_genesis_exists = genesis_checkup().await; if !local_genesis_exists || remote_height > local_height + 10 { imported_chain = true; info!("[sync] Starting sync from {local_height} to {remote_height}"); - node_syncing( + if let Err(err) = node_syncing( stream.clone(), ¶ms.db, remote_height, @@ -223,7 +288,31 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str current_key.clone(), ) .await - .map_err(|e| format!("Sync error: {e}"))?; + { + if err.kind() != io::ErrorKind::ConnectionAborted + && err.kind() != io::ErrorKind::BrokenPipe + && err.kind() != io::ErrorKind::ConnectionReset + && err.kind() != io::ErrorKind::TimedOut + { + return Err(format!("Sync error: {err}")); + } + + warn!( + "[sync] source peer {current_key} became unusable; preserving startup sync ownership and selecting another peer: {err}" + ); + let failed_stream = stream.clone(); + let current_stream = Connection::get_stream_from_memory(¤t_key).await; + if current_stream + .as_ref() + .map(|candidate| Arc::ptr_eq(candidate, &failed_stream)) + .unwrap_or(false) + { + remove_stream_from_memory(&failed_stream).await; + } + (current_key, stream) = wait_for_replacement_sync_peer(&failed_stream).await; + info!("[sync] resuming startup synchronization through {current_key}"); + continue; + } if !local_genesis_exists && !genesis_checkup().await && remote_height > 0 { return Err("Sync completed without obtaining remote genesis".to_string()); } @@ -235,57 +324,125 @@ pub async fn bootstrap_peer_discovery(params: BootstrapParams) -> Result<(), Str break; } - let post_sync_local_height = get_height(¶ms.db); - let post_sync_remote_height = - request_remote_height(stream.clone(), params.map.clone(), current_key.clone()).await?; - - let imported_candidates = match hydrate_torrent_candidates( - stream.clone(), - params.map.clone(), - current_key.clone(), - ) - .await - { - Ok(imported) => { - if imported > 0 { + let (post_sync_local_height, post_sync_remote_height) = loop { + let local_height = get_height(¶ms.db); + match request_remote_height(stream.clone(), params.map.clone(), current_key.clone()) + .await + { + Ok(remote_height) => break (local_height, remote_height), + Err(err) => { warn!( - "[sync] hydrated {imported} torrent candidates before post-sync orphan check" - ); + "[sync] source peer {current_key} failed post-sync height request; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = wait_for_replacement_sync_peer(&failed_stream).await; } - imported - } - Err(err) => { - warn!("[sync] failed to hydrate torrent candidates: {err}"); - 0 } }; - if post_sync_remote_height != post_sync_local_height || imported_candidates > 0 { - let orphan_checkup_params = OrphanCheckup2 { - stream: stream.clone(), - db: params.db.clone(), - local_height: post_sync_local_height, - remote_height: post_sync_remote_height, - recheck_from_height: Some(post_sync_local_height.min(post_sync_remote_height)), - map: params.map.clone(), - node_syncing: true, - connections_key: current_key.clone(), - }; - match sync_checkup(orphan_checkup_params, params.wallet.clone()).await { - Ok(()) => {} - Err(err) => return Err(format!("Post-sync orphan check error: {err}")), - } - } - - if imported_chain { - sync_governance_state( + let imported_candidates = loop { + match hydrate_torrent_candidates( stream.clone(), - ¶ms.db, params.map.clone(), current_key.clone(), ) .await - .map_err(|err| format!("Governance state sync error: {err}"))?; + { + Ok(imported) => { + if imported > 0 { + warn!( + "[sync] hydrated {imported} torrent candidates before post-sync orphan check" + ); + } + break imported; + } + Err(err) if is_peer_transport_failure(&err) => { + warn!( + "[sync] source peer {current_key} failed candidate hydration; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = wait_for_replacement_sync_peer(&failed_stream).await; + } + Err(err) => { + warn!("[sync] failed to hydrate torrent candidates: {err}"); + break 0; + } + } + }; + + if post_sync_remote_height != post_sync_local_height || imported_candidates > 0 { + loop { + let local_height = get_height(¶ms.db); + let remote_height = match request_remote_height( + stream.clone(), + params.map.clone(), + current_key.clone(), + ) + .await + { + Ok(height) => height, + Err(err) => { + warn!( + "[sync] post-sync source {current_key} failed height refresh; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = + wait_for_replacement_sync_peer(&failed_stream).await; + continue; + } + }; + let orphan_checkup_params = OrphanCheckup2 { + stream: stream.clone(), + db: params.db.clone(), + local_height, + remote_height, + recheck_from_height: Some(local_height.min(remote_height)), + map: params.map.clone(), + node_syncing: true, + connections_key: current_key.clone(), + }; + match sync_checkup(orphan_checkup_params, params.wallet.clone()).await { + Ok(()) => break, + Err(err) if is_peer_transport_failure(&err) => { + warn!( + "[sync] post-sync orphan source {current_key} disconnected; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = + wait_for_replacement_sync_peer(&failed_stream).await; + } + Err(err) => return Err(format!("Post-sync orphan check error: {err}")), + } + } + } + + if imported_chain { + loop { + match sync_governance_state( + stream.clone(), + ¶ms.db, + params.map.clone(), + current_key.clone(), + ) + .await + { + Ok(()) => break, + Err(err) if is_peer_transport_failure(&err) => { + warn!( + "[sync] source peer {current_key} failed governance synchronization; selecting another synchronized peer: {err}" + ); + let failed_stream = stream.clone(); + remove_stream_from_memory(&failed_stream).await; + (current_key, stream) = + wait_for_replacement_sync_peer(&failed_stream).await; + } + Err(err) => return Err(format!("Governance state sync error: {err}")), + } + } info!("[governance] finalized state sync completed: peer={current_key}"); } @@ -383,6 +540,18 @@ mod tests { drop(owner); assert!(try_start_startup_discovery().is_some()); } + + #[test] + fn transport_failures_are_separate_from_consensus_failures() { + assert!(is_peer_transport_failure( + "Timed out waiting for block hash vote at height 0" + )); + assert!(is_peer_transport_failure("Connection reset by peer")); + assert!(!is_peer_transport_failure( + "incompatible genesis from peer: local=a remote=b" + )); + assert!(!is_peer_transport_failure("Incorrect previous_block_hash")); + } } pub async fn process_handshake_response( diff --git a/src/rpc/server/connection_memory_manager.rs b/src/rpc/server/connection_memory_manager.rs index 9a02e08..e93d61f 100644 --- a/src/rpc/server/connection_memory_manager.rs +++ b/src/rpc/server/connection_memory_manager.rs @@ -1,8 +1,8 @@ use crate::common::binary_conversions::{binary_to_ip, ip_to_binary}; use crate::log::warn; use crate::records::memory::connections::{ - miner_connection_count, set_stream_health, spawn_outage_recovery_from_runtime, - spawn_retry_dropped_outgoing, CONNECTIONS, + preferred_miner_connection_type, set_stream_health, spawn_outage_recovery_from_runtime, + spawn_retry_dropped_outgoing, sync_capable_miner_connection_count, CONNECTIONS, }; use crate::records::memory::enums::{ClientType, ConnectionType}; use crate::records::memory::response_channels::{ @@ -90,7 +90,12 @@ async fn miner_stream_health( health } -async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) -> bool { +async fn purge_stale_duplicate_miner( + ip: &str, + new_connection_type: ConnectionType, + peer_wallet_short_address: &str, + command_map: Arc>, +) -> bool { let ip_bytes = ip_to_binary(ip); let duplicate = { let guard = CONNECTIONS.read().await; @@ -114,13 +119,20 @@ async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) connection_key.port, Arc::clone(&connection_info.stream), connection_info.health, + connection_info.ready, )) }) }) }; - let Some((connection_type, duplicate_ip, duplicate_port, duplicate_stream, recorded_health)) = - duplicate + let Some(( + connection_type, + duplicate_ip, + duplicate_port, + duplicate_stream, + recorded_health, + ready, + )) = duplicate else { return false; }; @@ -128,14 +140,30 @@ async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) let duplicate_key = format!("{duplicate_ip}:{duplicate_port}"); match recorded_health { ConnectionHealth::Connected | ConnectionHealth::Busy => { - let health = miner_stream_health( - &duplicate_key, - duplicate_stream.clone(), - command_map.clone(), - ) - .await; - if matches!(health, ConnectionHealth::Connected | ConnectionHealth::Busy) { - return false; + if !ready && connection_type != new_connection_type { + // Simultaneous cross-connections are resolved identically at + // both ends: the lexicographically smaller wallet owns the + // outgoing half of the one surviving TCP connection. + let Some(preferred_type) = + preferred_miner_connection_type(peer_wallet_short_address) + else { + return false; + }; + if connection_type == preferred_type || new_connection_type != preferred_type { + return false; + } + } else { + // Same-direction duplicates and established connections keep + // the original stale-stream probe used by reconnect recovery. + let health = miner_stream_health( + &duplicate_key, + duplicate_stream.clone(), + command_map.clone(), + ) + .await; + if matches!(health, ConnectionHealth::Connected | ConnectionHealth::Busy) { + return false; + } } } ConnectionHealth::Unresponsive | ConnectionHealth::Closed => {} @@ -201,7 +229,14 @@ pub async fn write_outgoing_miner_to_memory( return true; } - if !purge_stale_duplicate_miner(&ip, command_map.clone()).await { + if !purge_stale_duplicate_miner( + &ip, + ConnectionType::Outgoing, + &wallet_short_address, + command_map.clone(), + ) + .await + { return false; } @@ -263,7 +298,13 @@ pub async fn write_to_memory( drop(connection_instance); if !added && client_type == ClientType::Miner - && purge_stale_duplicate_miner(&ip, command_map.clone()).await + && purge_stale_duplicate_miner( + &ip, + ConnectionType::Incoming, + &wallet_short_address, + command_map.clone(), + ) + .await { let mut retry_connection_instance = CONNECTIONS.write().await; if let Some(mut retry_connection) = retry_connection_instance.take() { @@ -344,7 +385,7 @@ pub async fn remove_stream_from_memory(stream: &Arc>) { { spawn_retry_dropped_outgoing(ip.clone(), port); } - if miner_connection_count().await == 0 { + if sync_capable_miner_connection_count().await == 0 { spawn_outage_recovery_from_runtime(); } } diff --git a/src/rpc/server/flood_protection.rs b/src/rpc/server/flood_protection.rs index a5fc1de..54fe0b4 100644 --- a/src/rpc/server/flood_protection.rs +++ b/src/rpc/server/flood_protection.rs @@ -12,6 +12,28 @@ use crate::wallets::structures::Wallet; use crate::Arc; use crate::Utc; +fn is_bulk_mapping_record(client_type: ClientType, command: u8) -> bool { + client_type == ClientType::Miner + && matches!( + command, + RPC_NETWORK_MEMBERSHIP_RECONCILE | RPC_NETWORK_MONITOR_ADD | RPC_NETWORK_MONITOR_REMOVE + ) +} + +fn request_limit(client_type: ClientType, command: u8, mapping_records: usize) -> u32 { + if !is_bulk_mapping_record(client_type, command) { + return RPC_LONG_WINDOW_LIMIT; + } + // Allow an initial full reconciliation plus three setup retries, alongside + // the ordinary live-update allowance. Still score sustained repeated floods. + // Never sleep in the shared RPC reader: that would also stall block replies. + RPC_LONG_WINDOW_LIMIT.saturating_add( + u32::try_from(mapping_records) + .unwrap_or(u32::MAX) + .saturating_mul(4), + ) +} + pub const MAX_TORRENT_METADATA_BYTES: usize = 8192; pub const RPC_SHORT_WINDOW_SECS: i64 = 2; pub const RPC_LONG_WINDOW_SECS: i64 = 60; @@ -51,6 +73,12 @@ pub async fn check_request_frequency_with_client_type( command: u8, wallet: Arc, ) { + let mapping_records = if is_bulk_mapping_record(client_type, command) { + crate::records::memory::network_mapping::NodeInfo::mapping_sync_record_count().await + } else { + 0 + }; + let limit = request_limit(client_type, command, mapping_records); // Keep one compact flood-tracker row per subject and decay the // counters by elapsed time so stale request history expires // automatically without growing the tree unbounded. @@ -85,7 +113,7 @@ pub async fn check_request_frequency_with_client_type( // Only sustained long-window flooding is scored; short-window state // exists so the serialized tracker can be extended without changing // the storage shape again. - if state.is_flooding() { + if state.long_window_count > limit { let _ = update_ip_score( &ip, client_type.as_str(), @@ -108,6 +136,78 @@ mod tests { use super::*; use crate::rpc::command_maps::{RPC_TIME, RPC_VALIDATE_ADDRESS}; + #[test] + fn large_mapping_reconciliations_and_retries_fit_but_sustained_floods_do_not() { + for command in [ + RPC_NETWORK_MEMBERSHIP_RECONCILE, + RPC_NETWORK_MONITOR_ADD, + RPC_NETWORK_MONITOR_REMOVE, + ] { + let limit = request_limit(ClientType::Miner, command, 1_000); + let mut state = RpcFloodState::new(0); + for _ in 0..4_000 { + state.record_request(0); + } + assert!(state.long_window_count <= limit); + for _ in 0..=RPC_LONG_WINDOW_LIMIT { + state.record_request(0); + } + assert!(state.long_window_count > limit); + state.record_request(RPC_LONG_WINDOW_SECS); + assert!(state.long_window_count <= limit); + } + } + + #[test] + fn importing_a_map_from_empty_state_does_not_hit_the_old_250_record_limit() { + let mut state = RpcFloodState::new(0); + for already_imported in 0..1_000 { + state.record_request(0); + let limit = request_limit( + ClientType::Miner, + RPC_NETWORK_MEMBERSHIP_RECONCILE, + already_imported, + ); + assert!(state.long_window_count <= limit); + } + } + + #[test] + fn only_miner_bulk_records_get_scaled_limits() { + for command in [ + RPC_NETWORK_MEMBERSHIP_RECONCILE, + RPC_NETWORK_MONITOR_ADD, + RPC_NETWORK_MONITOR_REMOVE, + ] { + assert_eq!( + request_limit(ClientType::Miner, command, 0), + RPC_LONG_WINDOW_LIMIT + ); + assert_eq!( + request_limit(ClientType::Client, command, 100_000), + RPC_LONG_WINDOW_LIMIT + ); + } + for command in [ + RPC_TIME, + RPC_ADD_NETWORK_NODE, + RPC_SETUP_COMPLETE, + RPC_NETWORK_MAPPING_HASH, + ] { + assert_eq!( + request_limit(ClientType::Miner, command, 100_000), + RPC_LONG_WINDOW_LIMIT + ); + } + let mut state = RpcFloodState::new(0); + for _ in 0..RPC_LONG_WINDOW_LIMIT { + state.record_request(0); + } + assert!(!state.is_flooding()); + state.record_request(0); + assert!(state.is_flooding()); + } + #[test] fn miner_control_commands_have_independent_bounded_subjects() { let ip = "203.0.113.10"; diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index 0804741..933fe21 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -1,7 +1,7 @@ use crate::common::check_genesis::genesis_checkup; use crate::common::network_startup::canonical_miner_endpoint; use crate::log::{error, warn}; -use crate::miner::flag::{begin_chain_sync, ChainOperationGuard}; +use crate::miner::flag::{begin_connected_peer_sync, ChainOperationGuard}; use crate::orphans::structs::OrphanCheckup2; use crate::orphans::sync_check::sync_checkup; use crate::orphans::torrent_candidates::hydrate_torrent_candidates; @@ -17,6 +17,7 @@ use crate::records::memory::structs::Connection; use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::governance_state_sync::sync_governance_state; use crate::rpc::client::handshake::{connect_and_handshake, sponsor_endpoint_from_error}; +use crate::rpc::client::handshake_processing::startup_discovery_active; use crate::rpc::client::register_wallet::register_connected_wallet; use crate::rpc::client::structs::Connect; use crate::rpc::client::syncing::node_syncing; @@ -42,6 +43,86 @@ use crate::Settings; use crate::TcpStream; use crate::Utc; +#[cfg(test)] +mod stale_setup_reproduction { + use super::*; + use crate::miner::flag::{begin_chain_sync, is_mining_stop_requested, is_normal_mode, is_syncing_mode, set_mining_state, MiningState}; + use crate::records::memory::connections::CONNECTIONS; + use crate::records::memory::enums::ConnectionType; + use crate::records::memory::structs::{ConnectionInfo, ConnectionKey}; + use crate::rpc::command_maps::{RPC_BLOCK_HEIGHT, RPC_BLOCK_HASH_AT_HEIGHT}; + use crate::wallets::structures::SavedWallet; + use tokio::io::AsyncReadExt; + + // Regression for removed setup work taking the chain gate and stopping + // mining. No chain data is changed by this test. + #[tokio::test] + async fn disconnected_queued_setup_exits_without_stopping_mining() { + set_mining_state(MiningState::Idle); + let owner = begin_chain_sync().await; + let db = sled::Config::new().temporary(true).open().unwrap(); + let map = Arc::new(Mutex::new(Command::new())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()).await.unwrap(); + let (mut peer, _) = listener.accept().await.unwrap(); + let stream = Arc::new(Mutex::new(client)); + let key = ConnectionKey { connection_type: ConnectionType::Incoming.as_bytes(), ip: crate::common::binary_conversions::ip_to_binary("127.0.0.1"), port: 50050 }; + let info = ConnectionInfo::new(key.connection_type.clone(), key.ip.clone(), key.port, stream.clone(), ClientType::Miner.as_bytes(), "test-peer".into()); + let previous_connections = CONNECTIONS.write().await.replace(Connection { connection_map: std::collections::HashMap::from([(key, info)]) }); + let wallet = Arc::new(Wallet { saved: SavedWallet { short_address: String::new(), vanity_address: None, public_key: String::new(), private_key: String::new() }, encryption_key: String::new() }); + let mut setup = Box::pin(sync_incoming_peer_before_operational(stream.clone(), &db, wallet, map.clone(), "127.0.0.1:50050")); + let has_genesis = genesis_checkup().await; + let responder = async { + let mut request = [0u8; 4]; + peer.read_exact(&mut request).await.unwrap(); + assert_eq!(request[0], RPC_BLOCK_HEIGHT); + let uid = [request[1], request[2], request[3]]; + let tx = map.lock().await.get(&uid).unwrap().tx.clone(); + tx.send(0u32.to_le_bytes().to_vec()).await.unwrap(); + if has_genesis { + let mut request = [0u8; 8]; + peer.read_exact(&mut request).await.unwrap(); + assert_eq!(request[0], RPC_BLOCK_HASH_AT_HEIGHT); + let header = crate::records::unpack_block::unpack_header::load_block_header(0).await.unwrap(); + let uid = [request[1], request[2], request[3]]; + let tx = map.lock().await.get(&uid).unwrap().tx.clone(); + tx.send(crate::decode(header.hash().await).unwrap()).await.unwrap(); + } + }; + tokio::select! { + _ = &mut setup => panic!("setup exited before waiting for chain access"), + _ = responder => {} + } + // Poll through the initial checks into the occupied chain gate. + assert!(tokio::time::timeout(Duration::from_millis(50), &mut setup).await.is_err()); + remove_stream_from_memory(&stream).await; + assert!(CONNECTIONS.read().await.as_ref().unwrap().connection_map.is_empty()); + stream.lock().await.shutdown().await.unwrap(); + // Removed work must exit even while another operation owns the gate. + let result = tokio::time::timeout(Duration::from_secs(1), &mut setup).await.unwrap(); + assert!(matches!(result, Err(ref e) if e == "Incoming setup connection was removed while waiting for chain access")); + // Exiting stale work must not release the other operation's gate. + assert!(is_syncing_mode()); + assert!(is_mining_stop_requested()); + owner.finish(); + assert!(is_normal_mode()); + assert!(!is_mining_stop_requested()); + // Reconnecting at the same endpoint cannot make the old task current. + let replacement = Arc::new(Mutex::new(TcpStream::connect(listener.local_addr().unwrap()).await.unwrap())); + let (_replacement_peer, _) = listener.accept().await.unwrap(); + let key = ConnectionKey { connection_type: ConnectionType::Incoming.as_bytes(), ip: crate::common::binary_conversions::ip_to_binary("127.0.0.1"), port: 50050 }; + let info = ConnectionInfo::new(key.connection_type.clone(), key.ip.clone(), key.port, replacement.clone(), ClientType::Miner.as_bytes(), "test-peer".into()); + CONNECTIONS.write().await.as_mut().unwrap().connection_map.insert(key, info); + assert!(begin_connected_peer_sync(&stream).await.is_none()); + assert!(is_normal_mode()); + assert!(!is_mining_stop_requested()); + let valid_guard = begin_connected_peer_sync(&replacement).await.expect("current replacement must still be able to sync"); + assert!(is_syncing_mode()); + valid_guard.finish(); + *CONNECTIONS.write().await = previous_connections; + } +} + const LIVE_ORPHAN_WINDOW: u32 = 10; fn peer_reached_catch_up_target( @@ -114,7 +195,12 @@ async fn sync_incoming_peer_before_operational( // Wait for any existing startup, live catch-up, or orphan operation to // finish before this incoming peer can mutate the canonical chain. - let chain_sync_guard = begin_chain_sync().await; + let chain_sync_guard = begin_connected_peer_sync(&stream) + .await + .ok_or_else(|| "Incoming setup connection was removed while waiting for chain access".to_string())?; + if !incoming_setup_stream_is_current(connections_key, &stream).await { + return Err("Incoming setup connection was removed before chain synchronization".to_string()); + } let local_height = get_height(db); let remote_height = request_remote_height(stream.clone(), map.clone(), connections_key.to_string()).await?; @@ -398,7 +484,7 @@ async fn complete_incoming_miner_setup( db: db.clone(), wallet: wallet.clone(), map: map.clone(), - first: true, + first: !startup_discovery_active(), }; if let Err(sponsor_err) = connect_and_handshake(sponsor).await { error!( diff --git a/src/startup/connections.rs b/src/startup/connections.rs index eb6cd72..0de52f9 100644 --- a/src/startup/connections.rs +++ b/src/startup/connections.rs @@ -3,13 +3,14 @@ use crate::common::network_startup::{get_ip_and_port, get_node_connections}; use crate::log::{error, info, warn}; use crate::miner::flag::{is_mining_stop_requested, is_normal_mode}; use crate::records::memory::connections::{ - miner_connection_count, operational_peer_wallets, peer_connection_count, - ready_outgoing_connection_count, refill_outgoing_connections_once, + operational_peer_wallets, peer_connection_count, sync_capable_miner_connection_count, + // ready_outgoing_connection_count, refill_outgoing_connections_once, }; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; use crate::rpc::client::handshake::connect_and_handshake; use crate::rpc::client::handshake::outbound_handshake_in_progress; +use crate::rpc::client::handshake_processing::startup_discovery_active; use crate::rpc::client::structs::Connect; use crate::sled::Db; use crate::sleep; @@ -41,11 +42,6 @@ pub async fn handle_connections( wallet: Arc, map: Arc>, ) -> Result<(), String> { - // Reciprocal sponsorship is only the bootstrap rule for the first two - // nodes that are establishing a chain. A node restarting an existing - // chain may connect to any valid operational peer. - let first_two_node_startup = !genesis_checkup().await; - // A zero outgoing limit means this node waits for an incoming sponsor. It // does not bypass the independent-peer requirement or permit solo mining. let outgoing_connections = crate::Settings::load() @@ -68,6 +64,9 @@ pub async fn handle_connections( info!("No existing network peer is available. Waiting for connections."); let mut retry_seconds = 0_u64; loop { + // A reset node also starts without local genesis, but it stops being a + // first-two-node bootstrap as soon as synchronization imports genesis. + let first_two_node_startup = !genesis_checkup().await; let peer_wallets = operational_peer_wallets().await; let sponsorship_ready = if first_two_node_startup { NodeInfo::has_reciprocal_sponsorship(&wallet.saved.short_address, &peer_wallets).await @@ -105,18 +104,15 @@ async fn attempt_bootstrap_connections( ) -> Result { // Try the configured bootstrap peers one by one until a // handshake succeeds or the list is exhausted. - let mut filtered_servers = get_node_connections().await; - if context != "startup" { - // During a live outage the synchronized in-memory mapping is still - // available. Use its active endpoints as recovery candidates so the - // configured bootstrap host is not a runtime single point of failure. - for endpoint in NodeInfo::active_node_endpoints().await { - if !filtered_servers.contains(&endpoint) { - filtered_servers.push(endpoint); - } - } - } + let configured = get_node_connections().await; + let mapped = NodeInfo::active_node_endpoints().await; + let remembered = + crate::records::peer_endpoints::remembered_peer_endpoints(&db).unwrap_or_else(|err| { + warn!("[reconnect] could not read cached peer endpoints: {err}"); + Vec::new() + }); let (_, _, local_endpoint) = get_ip_and_port().await; + let filtered_servers = recovery_candidates(configured, mapped, remembered, &local_endpoint); let mut last_error: Option = None; for server in filtered_servers { @@ -136,7 +132,9 @@ async fn attempt_bootstrap_connections( // Clone the Arc for use in other async functions let map_clone = Arc::clone(&map); - let first: bool = true; + // An existing canonical startup owner will adopt this synchronized + // replacement stream. Do not queue a second full startup sync. + let first = !startup_discovery_active(); let connect_params = Connect { addr: socket_address, db: db_clone, @@ -200,6 +198,21 @@ async fn attempt_bootstrap_connections( Ok(false) } +fn recovery_candidates( + configured: Vec, + mapped: Vec, + remembered: Vec, + local_endpoint: &str, +) -> Vec { + let mut candidates = Vec::new(); + for endpoint in configured.into_iter().chain(mapped).chain(remembered) { + if endpoint != local_endpoint && !candidates.contains(&endpoint) { + candidates.push(endpoint); + } + } + candidates +} + pub fn spawn_outage_recovery(db: Db, wallet: Arc, map: Arc>) { let Some(recovery_guard) = try_start_outage_recovery() else { return; @@ -209,9 +222,9 @@ pub fn spawn_outage_recovery(db: Db, wallet: Arc, map: Arc, map: Arc, map: Arc usize { + block_number as usize % peer_count +} + fn expected_piece_hash( torrent: &crate::torrent::structs::Torrent, piece: u8, @@ -233,8 +237,12 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { } let mut reserved_any = false; + let peer_count = connected_nodes.len(); + let first_peer = first_peer_index(params.block_number, peer_count); for piece in &pieces { - for (connections_key, stream) in &connected_nodes { + for peer_offset in 0..peer_count { + let peer_index = (first_peer + peer_offset) % peer_count; + let (connections_key, stream) = &connected_nodes[peer_index]; // Use the full connection key for scheduling and failure // tracking so localhost peers on different ports remain distinct. let peer_key = connections_key.to_string(); @@ -327,3 +335,15 @@ pub async fn download_block_pieces(params: DownloadSave) -> Result<(), String> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::first_peer_index; + + #[test] + fn consecutive_blocks_rotate_across_available_peers() { + assert_eq!(first_peer_index(100, 3), 1); + assert_eq!(first_peer_index(101, 3), 2); + assert_eq!(first_peer_index(102, 3), 0); + } +} diff --git a/src/torrent/torrenting_system/get_nodes.rs b/src/torrent/torrenting_system/get_nodes.rs index b50a573..75367e3 100644 --- a/src/torrent/torrenting_system/get_nodes.rs +++ b/src/torrent/torrenting_system/get_nodes.rs @@ -1,6 +1,7 @@ use crate::common::binary_conversions::binary_to_ip; use crate::records::memory::connections::CONNECTIONS; use crate::records::memory::enums::ClientType; +use crate::records::memory::structs::ConnectionHealth; use crate::Arc; use crate::Mutex; use crate::TcpStream; @@ -18,6 +19,12 @@ pub async fn get_nodes_from_memory() -> Vec<(String, Arc>)> { if !connection_info.ready { continue; } + if !matches!( + connection_info.health, + ConnectionHealth::Connected | ConnectionHealth::Busy + ) { + continue; + } // Use ip:port as the scheduler key and clone the shared stream handle for requests. let ip = binary_to_ip(connection_info.ip.clone()); let port = connection_info.port; @@ -46,6 +53,12 @@ pub async fn get_sync_nodes_from_memory() -> Vec<(String, Arc>) { continue; } + if !matches!( + connection_info.health, + ConnectionHealth::Connected | ConnectionHealth::Busy + ) { + continue; + } let ip = binary_to_ip(connection_info.ip.clone()); let port = connection_info.port; let key = format!("{ip}:{port}");