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

recovery fixes

This commit is contained in:
contractless 2026-09-11 17:22:31 -06:00
parent 7ca01b9e69
commit 291fbd5755
17 changed files with 913 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/miner/flag.rs
View File

@ -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<crate::Mutex<crate::TcpStream>>,
) -> Option<ChainOperationGuard> {
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.

105
src/records/memory/connections.rs
View File

@ -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<NodeRuntimeContext> {
.and_then(|context| context.clone())
}
pub(crate) fn preferred_miner_connection_type(
peer_wallet_short_address: &str,
) -> Option<ConnectionType> {
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<Arc<Mutex<TcpStream>>> {
@ -835,6 +870,30 @@ impl Connection {
.collect()
}
pub fn get_sync_capable_peer_streams_with_keys(&self) -> Vec<(String, Arc<Mutex<TcpStream>>)> {
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<Arc<Mutex<TcpStream>>> {
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<Mutex<TcpStream>>) -> bool {
CONNECTIONS
.write()
@ -1616,6 +1690,15 @@ pub async fn startup_synced_peer_streams() -> Vec<(String, Arc<Mutex<TcpStream>>
.unwrap_or_default()
}
pub async fn sync_capable_peer_streams() -> Vec<(String, Arc<Mutex<TcpStream>>)> {
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<ClientType> {
// Recover the stored client role from the serialized connection key
// used throughout the RPC layer.

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

@ -21,10 +21,27 @@ fn signature_is_empty(signature: &str) -> bool {
}
impl NodeInfo {
fn new_node_admission_limit_reached(
address_map: &HashMap<String, NodeInfo>,
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<String, NodeInfo>,
operational_addresses: &HashSet<String>,
current_timestamp: u64,
excluded_address: &str,
) -> Option<String> {
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<String, NodeInfo>,
operational_addresses: &HashSet<String>,
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())
);
}

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

@ -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<u64> = 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)
}

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

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

13
src/records/memory/response_channels.rs
View File

@ -198,11 +198,22 @@ pub async fn delete_entry(map: Arc<Mutex<Command>>, key: Byte3) {
}
pub async fn retire_peer_entries(map: Arc<Mutex<Command>>, peer: &str) {
retire_peer_entries_created_before(map, peer, Instant::now()).await;
}
pub async fn retire_peer_entries_created_before(
map: Arc<Mutex<Command>>,
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;
}

1
src/records/mod.rs
View File

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

130
src/records/peer_endpoints.rs Normal file
View File

@ -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<String> {
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<String> = 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<Vec<String>> {
let previous: Vec<String> = 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::<u128>()));
{
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::<Vec<_>>();
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::<Vec<_>>();
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);
}
}

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

@ -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<Mutex<TcpStream>>,
) -> (String, Arc<Mutex<TcpStream>>) {
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(&params.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(),
&params.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(&current_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(&params.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(&params.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(),
&params.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(&params.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(),
&params.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(

73
src/rpc/server/connection_memory_manager.rs
View File

@ -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<Mutex<Command>>) -> bool {
async fn purge_stale_duplicate_miner(
ip: &str,
new_connection_type: ConnectionType,
peer_wallet_short_address: &str,
command_map: Arc<Mutex<Command>>,
) -> 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<Mutex<Command>>)
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<Mutex<Command>>)
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<Mutex<TcpStream>>) {
{
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();
}
}

102
src/rpc/server/flood_protection.rs
View File

@ -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<Wallet>,
) {
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";

92
src/rpc/server/handshake.rs
View File

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

81
src/startup/connections.rs
View File

@ -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<Wallet>,
map: Arc<Mutex<Command>>,
) -> 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<bool, String> {
// 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<String> = 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<String>,
mapped: Vec<String>,
remembered: Vec<String>,
local_endpoint: &str,
) -> Vec<String> {
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<Wallet>, map: Arc<Mutex<Command>>) {
let Some(recovery_guard) = try_start_outage_recovery() else {
return;
@ -209,9 +222,9 @@ pub fn spawn_outage_recovery(db: Db, wallet: Arc<Wallet>, map: Arc<Mutex<Command
let _recovery_guard = recovery_guard;
// A raw socket that never completed setup is not network recovery.
// Only an operational miner peer should suppress configured-peer
// outage recovery.
if outbound_handshake_in_progress() || miner_connection_count().await != 0 {
// Only a ready peer or one that completed registry and map sync can
// suppress configured-peer outage recovery.
if outbound_handshake_in_progress() || sync_capable_miner_connection_count().await != 0 {
return;
}
@ -248,9 +261,12 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc<Wallet>, map: Arc<M
// A topology pass is bounded: it walks the current active map once,
// tries each eligible endpoint at most once, then sleeps until the
// next interval even when the configured limit exceeds network size.
if ready_outgoing_connection_count().await < outgoing_connections as usize {
refill_outgoing_connections_once().await;
}
// Temporarily disabled with direct dropped-peer retries: while
// connected, do not rediscover and redial a peer that went away.
// Startup discovery and isolated recovery above remain enabled.
// if ready_outgoing_connection_count().await < outgoing_connections as usize {
// refill_outgoing_connections_once().await;
// }
}
});
}
@ -259,6 +275,21 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc<Wallet>, map: Arc<M
mod tests {
use super::*;
#[test]
fn restart_recovery_retains_cached_fallback_after_configured_peers() {
let peers = recovery_candidates(
vec!["8.8.8.8:50050".into()],
Vec::new(),
vec![
"8.8.8.8:50050".into(),
"1.1.1.1:50050".into(),
"9.9.9.9:50050".into(),
],
"9.9.9.9:50050",
);
assert_eq!(peers, vec!["8.8.8.8:50050", "1.1.1.1:50050"]);
}
#[test]
fn outage_recovery_has_one_owner() {
OUTAGE_RECOVERY_ACTIVE.store(false, Ordering::SeqCst);

2
src/startup/initialize_startup.rs
View File

@ -7,7 +7,6 @@ use crate::miner::flag::{
request_mining_stop, set_mining_state, set_node_mode, MiningState, NodeMode,
};
use crate::records::memory::connections::initialize_connection;
use crate::records::memory::network_mapping::NodeInfo;
use crate::wallets::structures::Wallet;
use crate::Path;
@ -69,7 +68,6 @@ pub async fn prepare_pre_wallet_startup() {
set_node_mode(NodeMode::Startup);
request_mining_stop();
set_mining_state(MiningState::Idle);
NodeInfo::initialize_runtime_start();
initialize_connection().await;
create_file_paths().await;
}

8
src/startup/network_broadcast.rs
View File

@ -515,6 +515,14 @@ async fn get_network_mapping_response(
NodeInfo::validate_synced_monitor_state(monitor_state, db, &expected_targets).await?;
NodeInfo::install_synced_snapshot(db, snapshot, validated_monitor_state).await?;
NodeInfo::restore_or_rebuild_mined_counts(db).await?;
// The snapshot has been validated. Persist only its active dial hints;
// startup still requires the complete normal handshake and join exchange.
if let Err(err) = crate::records::peer_endpoints::remember_peer_endpoints(
db,
&NodeInfo::active_node_endpoints().await,
) {
warn!("[network_map] could not cache discovered peer endpoints: {err}");
}
}
Ok(())

22
src/torrent/torrenting_system/download_pieces.rs
View File

@ -18,6 +18,10 @@ use crate::Arc;
use crate::Duration;
use crate::Mutex;
fn first_peer_index(block_number: u32, peer_count: usize) -> 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);
}
}

13
src/torrent/torrenting_system/get_nodes.rs
View File

@ -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<Mutex<TcpStream>>)> {
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<Mutex<TcpStream>>)
{
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}");