From 1808193100e96b9cd8fd39a5d5248334476ba1f4 Mon Sep 17 00:00:00 2001 From: contractless <00.pur.ple@gmail.com> Date: Fri, 28 Aug 2026 21:38:30 -0600 Subject: [PATCH] startup fixes --- src/records/memory/connections.rs | 10 +- src/records/memory/network_mapping/add.rs | 130 +++++++++++++++++- src/records/memory/network_mapping/monitor.rs | 30 +++- src/records/memory/network_mapping/queries.rs | 25 ++-- src/rpc/client/handshake.rs | 22 +++ src/rpc/client/handshake_processing.rs | 16 ++- src/rpc/server/handshake.rs | 32 ++++- src/startup/connections.rs | 27 ++-- src/startup/network_broadcast.rs | 28 ++-- 9 files changed, 261 insertions(+), 59 deletions(-) diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index b780de2..b97c0a5 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -992,12 +992,12 @@ fn spawn_monitor_activation( } tokio::spawn(async move { - // A reconnecting membership remains pending until an active sponsor's - // monitor-add arrives. Do not let that pending node author its own - // monitor event first; monitor validation correctly rejects it. + // During reciprocal startup both peers can hold freshly sponsored + // pending memberships. Permit either side to author its monitor-add so + // the two activation events cannot deadlock waiting on one another. let mapping_prerequisites = timeout(Duration::from_secs(30), async { - while !NodeInfo::is_active_address(&monitoring_address).await - || !NodeInfo::contains_address(&monitored_address).await + while !NodeInfo::contains_address(&monitored_address).await + || !NodeInfo::is_active_or_sponsored_pending(&monitoring_address).await { sleep(Duration::from_millis(100)).await; } diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 9180fb0..af3c1af 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -1,5 +1,7 @@ use super::*; -use crate::records::memory::connections::drop_miner_connections_for_wallet; +use crate::records::memory::connections::{ + drop_miner_connections_for_wallet, operational_peer_wallets, +}; use crate::records::memory::network_mapping::monitor::MONITOR_EVENT_STATE; use crate::records::memory::network_mapping::structs::SyncedNodeState; use crate::records::memory::response_channels::reserve_transient_entry_with_context; @@ -19,6 +21,48 @@ fn signature_is_empty(signature: &str) -> bool { } impl NodeInfo { + 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 + .iter() + .filter(|(address, node)| { + 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(); + candidates.sort_by(|(left_address, left), (right_address, right)| { + left.added_timestamp + .cmp(&right.added_timestamp) + .then_with(|| left_address.cmp(right_address)) + }); + candidates + .first() + .map(|(_, node)| format!("{}:{}", node.ip, node.port)) + } + + 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}")) + .unwrap_or_default() + } + fn validate_snapshot_structure(replacement: &HashMap) -> Result<(), String> { let mut active_ips = HashMap::::new(); @@ -792,6 +836,8 @@ impl NodeInfo { connections_key, } = params; let current_timestamp = Utc::now().timestamp_millis() as u64; + let operational_addresses: HashSet = + operational_peer_wallets().await.into_iter().collect(); if !is_public_network_address(&edit.ip) { return RpcResponse::Binary(b"Error: Invalid network address".to_vec()); } @@ -900,8 +946,14 @@ impl NodeInfo { .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}", + "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}", super::SELF_ADD_BLOCK, Self::self_add_limit_height(), ) @@ -909,10 +961,15 @@ impl NodeInfo { } let mined_count = signer_node.map(|node| node.blocks_mined).unwrap_or(0); if mined_count < 100 { - return RpcResponse::Binary( - b"Error: This address cannot add nodes. It must mined 100 blocks before adding new nodes to the network" - .to_vec(), + 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 mine 100 blocks before adding new nodes to the network{redirect}" + ).into_bytes()); } } @@ -1122,6 +1179,69 @@ 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, &[]); + rejected.blocks_mined = 250; + map.insert("rejected.cltc".to_string(), rejected); + + let mut newer = snapshot_node("1.2.3.6", ONE_HOUR_MILLIS, &[]); + newer.port = 50052; + newer.blocks_mined = 100; + map.insert("newer.cltc".to_string(), newer); + + let mut older = snapshot_node("1.2.3.5", 2, &[]); + older.port = 50051; + older.blocks_mined = 100; + map.insert("older.cltc".to_string(), older); + + assert_eq!( + 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()) + ); + } + + #[test] + fn sponsor_redirect_ignores_deleted_young_and_under_mined_nodes() { + let now = ONE_HOUR_MILLIS * 3; + let mut map = HashMap::new(); + + let mut deleted = snapshot_node("1.2.3.4", 1, &[]); + deleted.blocks_mined = 100; + 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 under_mined = snapshot_node("1.2.3.6", 1, &[]); + under_mined.blocks_mined = 99; + map.insert("under-mined.cltc".to_string(), under_mined); + + assert_eq!( + NodeInfo::eligible_sponsor_endpoint_from_map( + &map, + &HashSet::from([ + "deleted.cltc".to_string(), + "young.cltc".to_string(), + "under-mined.cltc".to_string(), + ]), + now, + "none.cltc", + ), + None + ); + } + #[test] fn complete_snapshot_allows_membership_before_monitor_setup_finishes() { let mut map = HashMap::new(); diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index 07bb1e4..59e4638 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -106,6 +106,24 @@ mod tests { assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending)); } + #[test] + fn only_freshly_sponsored_pending_membership_may_author_monitor_add() { + let address = "pending.cltc"; + let mut pending = node(&[]); + pending.added_timestamp = 300; + pending.deleted_timestamp = 300; + pending.deleted_block = 0; + + assert!(NodeInfo::can_pending_member_add_monitor(address, &pending)); + + pending.deleted_block = 50; + assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending)); + + pending.deleted_block = 0; + pending.added_by = address.to_string(); + assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending)); + } + #[test] fn signed_monitor_wire_record_round_trips() { let monitored = "ab13318c26250b048db92920a80a86127c933b0c.cltc".to_string(); @@ -287,6 +305,13 @@ impl NodeInfo { edit.action == MONITOR_ACTION_ADD && edit.modified_timestamp > node.deleted_timestamp } + pub(crate) fn can_pending_member_add_monitor(address: &str, node: &NodeInfo) -> bool { + node.deleted_timestamp > 0 + && node.deleted_timestamp == node.added_timestamp + && node.deleted_block == 0 + && node.added_by != address + } + pub(super) fn discard_old_target_ip_events_from( state: &mut HashMap, address: &str, @@ -393,7 +418,10 @@ impl NodeInfo { let monitoring = address_map .get(&edit.monitoring_address) .ok_or_else(|| "monitoring address not found".to_string())?; - if monitoring.deleted_timestamp > 0 { + if monitoring.deleted_timestamp > 0 + && !(edit.action == MONITOR_ACTION_ADD + && Self::can_pending_member_add_monitor(&edit.monitoring_address, monitoring)) + { return Err("deleted node cannot update monitor relationships".to_string()); } diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 404c270..0199473 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -47,6 +47,18 @@ impl NodeInfo { .unwrap_or(false) } + pub(crate) async fn is_active_or_sponsored_pending(address: &str) -> bool { + ADDRESS_MAP + .lock() + .await + .get(address) + .map(|node| { + node.deleted_timestamp == 0 + || Self::can_pending_member_add_monitor(address, node) + }) + .unwrap_or(false) + } + pub(crate) async fn contains_address(address: &str) -> bool { ADDRESS_MAP.lock().await.contains_key(address) } @@ -343,19 +355,6 @@ impl NodeInfo { .collect() } - pub async fn eligible_sponsor_ips() -> Vec { - let current_timestamp = Utc::now().timestamp_millis() as u64; - let map = ADDRESS_MAP.lock().await; - map.values() - .filter(|node_info| { - node_info.deleted_timestamp == 0 - && current_timestamp.saturating_sub(node_info.added_timestamp) >= 3_600_000 - && node_info.blocks_mined >= 100 - }) - .map(|node_info| node_info.ip.clone()) - .collect() - } - pub async fn get_deleted_addresses() -> Vec { let map = ADDRESS_MAP.lock().await; map.iter() diff --git a/src/rpc/client/handshake.rs b/src/rpc/client/handshake.rs index 62f56c6..ff94f67 100644 --- a/src/rpc/client/handshake.rs +++ b/src/rpc/client/handshake.rs @@ -98,6 +98,28 @@ fn sponsor_endpoint_from_error(error: &str) -> Option { Some(endpoint.to_string()) } +#[cfg(test)] +mod tests { + use super::sponsor_endpoint_from_error; + + #[test] + fn parses_server_sponsor_redirect() { + let error = "network self-announcement was rejected: Error: sponsor unavailable sponsor=1.2.3.4:50050"; + assert_eq!( + sponsor_endpoint_from_error(error), + Some("1.2.3.4:50050".to_string()) + ); + } + + #[test] + fn ignores_rejection_without_sponsor_redirect() { + assert_eq!( + sponsor_endpoint_from_error("network self-announcement was rejected"), + None + ); + } +} + async fn connect_from_configured_ip(remote_addr: SocketAddr) -> io::Result { let listen_ip = get_listen_ip().await; let local_ip: IpAddr = listen_ip.parse().map_err(|err| { diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 1375b63..ea6a525 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -15,7 +15,7 @@ use crate::records::memory::connections::{ spawn_peer_setup_retry, startup_synced_peer_streams, CONNECTIONS, }; use crate::records::memory::network_mapping::NodeInfo; -use crate::records::memory::response_channels::Command; +use crate::records::memory::response_channels::{retire_peer_entries, Command}; 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; @@ -85,8 +85,15 @@ pub fn spawn_bootstrap_peer_discovery(params: BootstrapParams) { tokio::spawn(async move { let _startup_guard = startup_guard; + let failed_stream = params.stream.clone(); + let failed_key = params.connections_key.clone(); + let command_map = params.map.clone(); if let Err(e) = bootstrap_peer_discovery(params).await { - eprintln!("[bootstrap] error: {e}"); + warn!("[bootstrap] asynchronous peer setup failed for {failed_key}: {e}"); + remove_stream_from_memory(&failed_stream).await; + // The stream may already have been removed by the failing setup + // path. Retire its request waiters explicitly in either case. + retire_peer_entries(command_map, &failed_key).await; } }); } @@ -558,13 +565,10 @@ pub async fn process_handshake_response( .map_err(|err| io::Error::other(format!("Network-map digest comparison failed: {err}")))?; if !matched { - info!( - "[network_map] reconciling signed mapping state with peer {connections_key}" - ); + info!("[network_map] reconciling signed mapping state with peer {connections_key}"); reconcile_network_mapping_with_peer( broadcast_stream.clone(), params.map.clone(), - ¶ms.db, &connections_key, ) .await diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index ba68ab1..96a9e35 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -251,9 +251,27 @@ fn spawn_incoming_peer_promotion_watcher( }); } +fn needs_reverse_bootstrap_announcement( + self_add_allowed: bool, + local_membership_active: bool, +) -> bool { + self_add_allowed && !local_membership_active +} + #[cfg(test)] mod tests { - use super::{passive_catch_up_expired, peer_reached_catch_up_target}; + use super::{ + needs_reverse_bootstrap_announcement, passive_catch_up_expired, + peer_reached_catch_up_target, + }; + + #[test] + fn reverse_bootstrap_announcement_is_only_for_an_unregistered_first_node() { + assert!(needs_reverse_bootstrap_announcement(true, false)); + assert!(!needs_reverse_bootstrap_announcement(true, true)); + assert!(!needs_reverse_bootstrap_announcement(false, false)); + assert!(!needs_reverse_bootstrap_announcement(false, true)); + } #[test] fn moving_tip_does_not_move_the_original_catch_up_target() { @@ -357,12 +375,16 @@ async fn complete_incoming_miner_setup( crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( remote_height, ); + let short_address = wallet.saved.short_address.clone(); + let local_membership_active = + crate::records::memory::network_mapping::NodeInfo::is_active_address(&short_address).await; - if self_add_allowed { + if needs_reverse_bootstrap_announcement(self_add_allowed, local_membership_active) { // Before the sponsored-add gate, reverse announcement lets tiny early - // networks bootstrap from one live peer. After the gate activates, the - // incoming peer's outgoing join request is the only add path. - let short_address = wallet.saved.short_address.clone(); + // networks bootstrap from one live peer. It is only valid while this + // receiving node has no active membership of its own. An established + // node must never import even its own record from a newly connected + // peer. if let Err(err) = announce_self_to_network( stream.clone(), &short_address, diff --git a/src/startup/connections.rs b/src/startup/connections.rs index c981d25..fee6787 100644 --- a/src/startup/connections.rs +++ b/src/startup/connections.rs @@ -1,3 +1,4 @@ +use crate::common::check_genesis::genesis_checkup; 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}; @@ -39,6 +40,11 @@ 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() @@ -56,17 +62,22 @@ pub async fn handle_connections( // Startup gets one bounded pass through configured peers. If none can // sponsor this node, wait for a real incoming peer instead of repeatedly - // dialing the list. Genesis remains blocked until reciprocal sponsorship - // makes that peer operational. + // dialing the list. Only the first two nodes establishing genesis require + // reciprocal sponsorship; mature restarts need an operational peer. info!("No existing network peer is available. Waiting for connections."); loop { let peer_wallets = operational_peer_wallets().await; - if is_normal_mode() - && !is_mining_stop_requested() - && NodeInfo::has_reciprocal_sponsorship(&wallet.saved.short_address, &peer_wallets) - .await - { - info!("Reciprocal node sponsorship completed; startup may continue."); + let sponsorship_ready = if first_two_node_startup { + NodeInfo::has_reciprocal_sponsorship(&wallet.saved.short_address, &peer_wallets).await + } else { + !peer_wallets.is_empty() + }; + if is_normal_mode() && !is_mining_stop_requested() && sponsorship_ready { + if first_two_node_startup { + info!("Reciprocal node sponsorship completed; startup may continue."); + } else { + info!("Operational peer connected; existing-chain startup may continue."); + } return Ok(()); } sleep(Duration::from_secs(1)).await; diff --git a/src/startup/network_broadcast.rs b/src/startup/network_broadcast.rs index 0fe16e6..c0dc21a 100644 --- a/src/startup/network_broadcast.rs +++ b/src/startup/network_broadcast.rs @@ -140,15 +140,6 @@ async fn announce_self_to_network_inner( Ok(()) } -pub async fn get_network_mapping( - unlocked_stream: Arc>, - command_map: Arc>, - db: &Db, - connections_key: &str, -) -> Result<(), String> { - get_network_mapping_inner(unlocked_stream, command_map, db, connections_key, None).await -} - pub async fn compare_network_mapping_digest( unlocked_stream: Arc>, command_map: Arc>, @@ -196,7 +187,9 @@ async fn send_mapping_record( .ok_or_else(|| "network-map reconciliation response channel closed".to_string())?; let response = binary_to_string(response); if response != "Success" { - return Err(format!("peer rejected reconciled mapping record: {response}")); + return Err(format!( + "peer rejected reconciled mapping record: {response}" + )); } Ok(()) } @@ -204,7 +197,6 @@ async fn send_mapping_record( pub async fn reconcile_network_mapping_with_peer( unlocked_stream: Arc>, command_map: Arc>, - db: &Db, connections_key: &str, ) -> Result<(), String> { let (memberships, monitor_events) = NodeInfo::signed_mapping_state().await; @@ -218,7 +210,9 @@ pub async fn reconcile_network_mapping_with_peer( let signature = crate::decode(&edit.modified_signature) .map_err(|_| "local mapping contained an invalid membership signature".to_string())?; if signature.len() != Wallet::SIGNATURE_LENGTH { - return Err("local mapping contained an invalid membership signature length".to_string()); + return Err( + "local mapping contained an invalid membership signature length".to_string(), + ); } let mut body = Vec::new(); @@ -279,10 +273,12 @@ pub async fn reconcile_network_mapping_with_peer( .await?; } - // The peer has now merged every valid current signed record we hold. Pull - // its resulting complete state so records that only it possessed are also - // installed locally and derived deletion cascades are identical. - get_network_mapping(unlocked_stream, command_map, db, connections_key).await + // Reconciliation on an additional/discovered connection is intentionally + // one-way. This node already joined through its bootstrap owner, so the + // newly connected peer may merge our signed state but must never replace + // this node's established mapping. Missing records continue to arrive + // through the normal redundant network broadcasts. + Ok(()) } pub async fn get_network_mapping_for_address(