diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 45d1540..b780de2 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -171,7 +171,10 @@ async fn reconnect_replacement_inner(excluded_ip: &str) { }; let Some((connections_key, stream)) = live_connection else { - warn!("[reconnect] no live stream available for bootstrap recovery"); + // Topology discovery needs an operational stream. If none remains, + // fall back to configured-peer outage recovery instead of retrying + // this impossible operation every minute. + spawn_outage_recovery(context.db, context.wallet, context.map); return; }; @@ -203,7 +206,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 peer_connection_count().await == 0 { spawn_outage_recovery(context.db, context.wallet, context.map); return; } @@ -992,16 +995,18 @@ fn spawn_monitor_activation( // 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. - let local_activation = timeout(Duration::from_secs(30), async { - while !NodeInfo::is_active_address(&monitoring_address).await { + let mapping_prerequisites = timeout(Duration::from_secs(30), async { + while !NodeInfo::is_active_address(&monitoring_address).await + || !NodeInfo::contains_address(&monitored_address).await + { sleep(Duration::from_millis(100)).await; } }) .await; - if local_activation.is_err() { + if mapping_prerequisites.is_err() { warn!( - "[network_map] connection setup timed out waiting for local membership activation: address={} peer={}:{}", - monitoring_address, ip, port + "[network_map] connection setup timed out waiting for mapping prerequisites: local={} monitored={} peer={}:{}", + monitoring_address, monitored_address, ip, port ); remove_stream_from_memory(&stream).await; return; diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 6096604..9180fb0 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -208,41 +208,56 @@ impl NodeInfo { pub(crate) async fn finish_mapping_snapshot_sync() { loop { - let update = { + let updates = { let mut state = MAPPING_SNAPSHOT_STATE.lock().await; - match state.updates.pop_front() { - Some(update) => Some(update), - None => { - state.active = false; - None - } + if state.updates.is_empty() { + state.active = false; + Vec::new() + } else { + state.updates.drain(..).collect::>() } }; - match update { - Some(DeferredMappingUpdate::Add(db, edit)) => { - if let Err(err) = Self::import_signed_mapping_address(&db, edit, 0).await { - warn!("[network_map] deferred node update was rejected: {err}"); + if updates.is_empty() { + break; + } + + // Membership is the prerequisite for monitor validation. Live + // broadcasts can reach a syncing node in either order, so replay + // every deferred membership/snapshot change before monitor events. + let mut monitor_updates = Vec::new(); + for update in updates { + match update { + DeferredMappingUpdate::Monitor(params, action) => { + monitor_updates.push((params, action)); + } + DeferredMappingUpdate::Add(db, edit) => { + if let Err(err) = Self::import_signed_mapping_address(&db, edit, 0).await { + warn!("[network_map] deferred node update was rejected: {err}"); + } + } + DeferredMappingUpdate::SnapshotAddress(db, state) => { + if let Err(err) = Self::import_synced_mapping_address_now(&db, state).await + { + warn!("[network_map] deferred snapshot-address update was rejected: {err}"); + } + } + DeferredMappingUpdate::ReconciledMembership(db, state) => { + if let Err(err) = Self::import_reconciled_membership_now(&db, state).await { + warn!( + "[network_map] deferred reconciled membership was rejected: {err}" + ); + } } } - Some(DeferredMappingUpdate::Monitor(params, action)) => { - let RpcResponse::Binary(bytes) = Self::apply_monitor_now(params, action).await; - let response = String::from_utf8_lossy(&bytes); - if response != "Success" { - warn!("[network_map] deferred monitor update was rejected: {response}"); - } + } + + for (params, action) in monitor_updates { + let RpcResponse::Binary(bytes) = Self::apply_monitor_now(params, action).await; + let response = String::from_utf8_lossy(&bytes); + if response != "Success" { + warn!("[network_map] deferred monitor update was rejected: {response}"); } - Some(DeferredMappingUpdate::SnapshotAddress(db, state)) => { - if let Err(err) = Self::import_synced_mapping_address_now(&db, state).await { - warn!("[network_map] deferred snapshot-address update was rejected: {err}"); - } - } - Some(DeferredMappingUpdate::ReconciledMembership(db, state)) => { - if let Err(err) = Self::import_reconciled_membership_now(&db, state).await { - warn!("[network_map] deferred reconciled membership was rejected: {err}"); - } - } - None => break, } } } diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 9af66ff..404c270 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -47,6 +47,10 @@ impl NodeInfo { .unwrap_or(false) } + pub(crate) async fn contains_address(address: &str) -> bool { + ADDRESS_MAP.lock().await.contains_key(address) + } + pub(crate) async fn signed_mapping_state() -> (Vec, Vec) { let map = ADDRESS_MAP.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await; diff --git a/src/startup/connections.rs b/src/startup/connections.rs index 4bfe518..c981d25 100644 --- a/src/startup/connections.rs +++ b/src/startup/connections.rs @@ -2,7 +2,7 @@ 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, ready_outgoing_connection_count, + operational_peer_wallets, peer_connection_count, ready_outgoing_connection_count, refill_outgoing_connections_once, }; use crate::records::memory::network_mapping::NodeInfo; @@ -48,28 +48,28 @@ pub async fn handle_connections( info!("OUTGOING_CONNECTIONS is 0; waiting for an incoming sponsored peer."); } + if outgoing_connections > 0 + && attempt_bootstrap_connections(db.clone(), wallet.clone(), map.clone(), "startup").await? + { + return Ok(()); + } + + // 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. + info!("No existing network peer is available. Waiting for connections."); loop { - if outgoing_connections > 0 - && attempt_bootstrap_connections(db.clone(), wallet.clone(), map.clone(), "startup") - .await? + 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."); return Ok(()); } - - info!("No existing network peer is available. Waiting for connections."); - for _ in 0..60 { - 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."); - return Ok(()); - } - sleep(Duration::from_secs(1)).await; - } - warn!("No sponsored peer is operational; retrying configured bootstrap peers"); + sleep(Duration::from_secs(1)).await; } } @@ -81,7 +81,17 @@ async fn attempt_bootstrap_connections( ) -> Result { // Try the configured bootstrap peers one by one until a // handshake succeeds or the list is exhausted. - let filtered_servers = get_node_connections().await; + 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 (_, _, local_endpoint) = get_ip_and_port().await; let mut last_error: Option = None; @@ -174,9 +184,10 @@ pub fn spawn_outage_recovery(db: Db, wallet: Arc, map: Arc, map: Arc