networkk connectivity fixes

This commit is contained in:
contractless 2026-08-21 14:01:10 -06:00
parent a370c18825
commit ff67566944
4 changed files with 94 additions and 59 deletions

View File

@ -171,7 +171,10 @@ async fn reconnect_replacement_inner(excluded_ip: &str) {
}; };
let Some((connections_key, stream)) = live_connection else { 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; 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 // Losing every miner peer is process-level outage recovery, not an
// independent retry of each socket that happened to time out. // 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); spawn_outage_recovery(context.db, context.wallet, context.map);
return; return;
} }
@ -992,16 +995,18 @@ fn spawn_monitor_activation(
// A reconnecting membership remains pending until an active sponsor's // A reconnecting membership remains pending until an active sponsor's
// monitor-add arrives. Do not let that pending node author its own // monitor-add arrives. Do not let that pending node author its own
// monitor event first; monitor validation correctly rejects it. // monitor event first; monitor validation correctly rejects it.
let local_activation = timeout(Duration::from_secs(30), async { let mapping_prerequisites = timeout(Duration::from_secs(30), async {
while !NodeInfo::is_active_address(&monitoring_address).await { while !NodeInfo::is_active_address(&monitoring_address).await
|| !NodeInfo::contains_address(&monitored_address).await
{
sleep(Duration::from_millis(100)).await; sleep(Duration::from_millis(100)).await;
} }
}) })
.await; .await;
if local_activation.is_err() { if mapping_prerequisites.is_err() {
warn!( warn!(
"[network_map] connection setup timed out waiting for local membership activation: address={} peer={}:{}", "[network_map] connection setup timed out waiting for mapping prerequisites: local={} monitored={} peer={}:{}",
monitoring_address, ip, port monitoring_address, monitored_address, ip, port
); );
remove_stream_from_memory(&stream).await; remove_stream_from_memory(&stream).await;
return; return;

View File

@ -208,41 +208,56 @@ impl NodeInfo {
pub(crate) async fn finish_mapping_snapshot_sync() { pub(crate) async fn finish_mapping_snapshot_sync() {
loop { loop {
let update = { let updates = {
let mut state = MAPPING_SNAPSHOT_STATE.lock().await; let mut state = MAPPING_SNAPSHOT_STATE.lock().await;
match state.updates.pop_front() { if state.updates.is_empty() {
Some(update) => Some(update), state.active = false;
None => { Vec::new()
state.active = false; } else {
None state.updates.drain(..).collect::<Vec<_>>()
}
} }
}; };
match update { if updates.is_empty() {
Some(DeferredMappingUpdate::Add(db, edit)) => { break;
if let Err(err) = Self::import_signed_mapping_address(&db, edit, 0).await { }
warn!("[network_map] deferred node update was rejected: {err}");
// 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); for (params, action) in monitor_updates {
if response != "Success" { let RpcResponse::Binary(bytes) = Self::apply_monitor_now(params, action).await;
warn!("[network_map] deferred monitor update was rejected: {response}"); 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,
} }
} }
} }

View File

@ -47,6 +47,10 @@ impl NodeInfo {
.unwrap_or(false) .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<SyncedNodeState>, Vec<SignedMonitorEdit>) { pub(crate) async fn signed_mapping_state() -> (Vec<SyncedNodeState>, Vec<SignedMonitorEdit>) {
let map = ADDRESS_MAP.lock().await; let map = ADDRESS_MAP.lock().await;
let monitor_events = MONITOR_EVENT_STATE.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await;

View File

@ -2,7 +2,7 @@ use crate::common::network_startup::{get_ip_and_port, get_node_connections};
use crate::log::{error, info, warn}; use crate::log::{error, info, warn};
use crate::miner::flag::{is_mining_stop_requested, is_normal_mode}; use crate::miner::flag::{is_mining_stop_requested, is_normal_mode};
use crate::records::memory::connections::{ 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, refill_outgoing_connections_once,
}; };
use crate::records::memory::network_mapping::NodeInfo; 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."); 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 { loop {
if outgoing_connections > 0 let peer_wallets = operational_peer_wallets().await;
&& attempt_bootstrap_connections(db.clone(), wallet.clone(), map.clone(), "startup") if is_normal_mode()
.await? && !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(()); return Ok(());
} }
sleep(Duration::from_secs(1)).await;
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");
} }
} }
@ -81,7 +81,17 @@ async fn attempt_bootstrap_connections(
) -> Result<bool, String> { ) -> Result<bool, String> {
// Try the configured bootstrap peers one by one until a // Try the configured bootstrap peers one by one until a
// handshake succeeds or the list is exhausted. // 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 (_, _, local_endpoint) = get_ip_and_port().await;
let mut last_error: Option<String> = None; let mut last_error: Option<String> = None;
@ -174,9 +184,10 @@ pub fn spawn_outage_recovery(db: Db, wallet: Arc<Wallet>, map: Arc<Mutex<Command
tokio::spawn(async move { tokio::spawn(async move {
let _recovery_guard = recovery_guard; let _recovery_guard = recovery_guard;
// A miner socket still completing setup owns recovery even though it // A raw socket that never completed setup is not network recovery.
// is not ready for normal traffic yet. // Only an operational miner peer should suppress configured-peer
if miner_connection_count().await != 0 { // outage recovery.
if peer_connection_count().await != 0 {
return; return;
} }
@ -205,7 +216,7 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc<Wallet>, map: Arc<M
continue; continue;
} }
if miner_connection_count().await == 0 { if peer_connection_count().await == 0 {
spawn_outage_recovery(db.clone(), wallet.clone(), map.clone()); spawn_outage_recovery(db.clone(), wallet.clone(), map.clone());
continue; continue;
} }