From a370c188250efd15bcd87a31ce20f08f69851eed Mon Sep 17 00:00:00 2001 From: contractless <00.pur.ple@gmail.com> Date: Sun, 16 Aug 2026 15:17:18 -0600 Subject: [PATCH] network mapping reconnection fixes --- src/bin/lookup_online_node_count.rs | 175 ++++++++++++++ src/records/memory/connections.rs | 222 +++++++++++++----- src/records/memory/network_mapping/add.rs | 61 ++++- src/records/memory/network_mapping/monitor.rs | 39 ++- src/records/memory/network_mapping/queries.rs | 42 +++- src/records/memory/response_channels.rs | 43 ++++ src/records/memory/structs.rs | 2 + src/rpc/server/connection_memory_manager.rs | 22 +- src/startup/connections.rs | 73 ++++-- 9 files changed, 584 insertions(+), 95 deletions(-) create mode 100644 src/bin/lookup_online_node_count.rs diff --git a/src/bin/lookup_online_node_count.rs b/src/bin/lookup_online_node_count.rs new file mode 100644 index 0000000..c760104 --- /dev/null +++ b/src/bin/lookup_online_node_count.rs @@ -0,0 +1,175 @@ +use contractless::common::network_startup::get_connections; +use contractless::env; +use contractless::records::memory::network_mapping::structs::{ + NETWORK_SNAPSHOT_HEADER_BYTES, NETWORK_SNAPSHOT_MAGIC, NETWORK_SNAPSHOT_VERSION, + NODE_DELETED_TIMESTAMP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_RECORD_FIXED_BYTES, +}; +use contractless::records::memory::response_channels::generate_uid; +use contractless::standalone_tools::connections::handshake; +use contractless::wallets::structures::Wallet; + +fn online_node_count(response: &[u8]) -> Result { + if response.len() < NETWORK_SNAPSHOT_HEADER_BYTES { + return Err("network mapping response was shorter than its header".to_string()); + } + if &response[..NETWORK_SNAPSHOT_MAGIC.len()] != NETWORK_SNAPSHOT_MAGIC { + return Err("network mapping response used an unknown format".to_string()); + } + if response[NETWORK_SNAPSHOT_MAGIC.len()] != NETWORK_SNAPSHOT_VERSION { + return Err("network mapping response used an unsupported version".to_string()); + } + + let mapping_len = u32::from_le_bytes( + response[5..9] + .try_into() + .map_err(|_| "network mapping length was invalid")?, + ) as usize; + let monitor_state_len = u32::from_le_bytes( + response[9..13] + .try_into() + .map_err(|_| "network monitor-state length was invalid")?, + ) as usize; + let mapping_end = NETWORK_SNAPSHOT_HEADER_BYTES + .checked_add(mapping_len) + .ok_or_else(|| "network mapping length overflowed".to_string())?; + let response_end = mapping_end + .checked_add(monitor_state_len) + .ok_or_else(|| "network monitor-state length overflowed".to_string())?; + if response_end != response.len() { + return Err("network mapping response length did not match its header".to_string()); + } + + let mut offset = NETWORK_SNAPSHOT_HEADER_BYTES; + let mut online = 0usize; + while offset < mapping_end { + let fixed_end = offset + .checked_add(NODE_RECORD_FIXED_BYTES) + .ok_or_else(|| "network node record length overflowed".to_string())?; + if fixed_end > mapping_end { + return Err("network mapping ended inside a node record".to_string()); + } + + let deleted_timestamp = u64::from_le_bytes( + response[offset + NODE_DELETED_TIMESTAMP_OFFSET + ..offset + NODE_DELETED_TIMESTAMP_OFFSET + 8] + .try_into() + .map_err(|_| "node deletion timestamp was invalid")?, + ); + let monitor_count = u16::from_le_bytes( + response[offset + NODE_MONITOR_COUNT_OFFSET + ..offset + NODE_MONITOR_COUNT_OFFSET + 2] + .try_into() + .map_err(|_| "node monitor count was invalid")?, + ) as usize; + let monitor_bytes = monitor_count + .checked_mul(Wallet::SHORT_ADDRESS_BYTES_LENGTH) + .ok_or_else(|| "node monitor list length overflowed".to_string())?; + let record_end = fixed_end + .checked_add(monitor_bytes) + .ok_or_else(|| "network node record length overflowed".to_string())?; + if record_end > mapping_end { + return Err("network mapping ended inside a node monitor list".to_string()); + } + + if deleted_timestamp == 0 && monitor_count > 0 { + online += 1; + } + offset = record_end; + } + + Ok(online) +} + +#[tokio::main] +async fn main() { + let args: Vec = env::args().collect(); + if args.len() != 3 { + eprintln!("Usage: lookup_online_node_count WALLET_PATH WALLET_KEY"); + std::process::exit(2); + } + + let wallet_path = args[1].clone(); + let encryption_key = args[2].clone(); + let rpc_command = 30; + + for connection in get_connections().await { + let socket_address = match connection.parse() { + Ok(address) => address, + Err(err) => { + eprintln!("Skipping invalid configured node {connection}: {err}"); + continue; + } + }; + let response = handshake::connect_and_handshake( + socket_address, + String::new(), + rpc_command, + handshake::HandshakeWallet::WalletKey { + encryption_key: encryption_key.clone(), + wallet_path: wallet_path.clone(), + }, + generate_uid(), + ) + .await; + + match response { + Ok(bytes) => match online_node_count(&bytes) { + Ok(count) => { + println!("{count}"); + return; + } + Err(err) => eprintln!("Invalid mapping from {connection}: {err}"), + }, + Err(err) => eprintln!("Failed to query {connection}: {err}"), + } + } + + eprintln!("failed to retrieve an online node count"); + std::process::exit(1); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn snapshot_record(deleted_timestamp: u64, monitors: u16) -> Vec { + let mut record = vec![0u8; NODE_RECORD_FIXED_BYTES]; + record[NODE_DELETED_TIMESTAMP_OFFSET..NODE_DELETED_TIMESTAMP_OFFSET + 8] + .copy_from_slice(&deleted_timestamp.to_le_bytes()); + record[NODE_MONITOR_COUNT_OFFSET..NODE_MONITOR_COUNT_OFFSET + 2] + .copy_from_slice(&monitors.to_le_bytes()); + record.extend(vec![ + 0u8; + monitors as usize * Wallet::SHORT_ADDRESS_BYTES_LENGTH + ]); + record + } + + fn snapshot(records: &[Vec]) -> Vec { + let mapping = records.concat(); + let mut response = Vec::new(); + response.extend_from_slice(NETWORK_SNAPSHOT_MAGIC); + response.push(NETWORK_SNAPSHOT_VERSION); + response.extend_from_slice(&(mapping.len() as u32).to_le_bytes()); + response.extend_from_slice(&0u32.to_le_bytes()); + response.extend_from_slice(&mapping); + response + } + + #[test] + fn counts_only_active_monitored_nodes() { + let response = snapshot(&[ + snapshot_record(0, 2), + snapshot_record(0, 0), + snapshot_record(1234, 0), + ]); + assert_eq!(online_node_count(&response).unwrap(), 1); + } + + #[test] + fn rejects_truncated_monitor_lists() { + let mut response = snapshot(&[snapshot_record(0, 1)]); + response.pop(); + assert!(online_node_count(&response).is_err()); + } +} diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 177ada1..45d1540 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -6,7 +6,7 @@ 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, Command, }; use crate::records::memory::structs::{Connection, ConnectionHealth, StoreConnectionParams}; use crate::rpc::client::handshake::connect_and_handshake; @@ -14,8 +14,10 @@ use crate::rpc::client::handshake_processing::{bootstrap_peer_discovery, Bootstr use crate::rpc::client::structs::Connect; use crate::rpc::command_maps::{RPC_BLOCK_HEIGHT, RPC_SETUP_COMPLETE}; use crate::rpc::responses::RpcResponse; +use crate::rpc::server::connection_memory_manager::remove_stream_from_memory; use crate::sled::Db; use crate::sleep; +use crate::startup::connections::spawn_outage_recovery; use crate::thread_rng; use crate::timeout; use crate::wallets::structures::Wallet; @@ -122,6 +124,20 @@ fn node_runtime_context() -> Option { .and_then(|context| context.clone()) } +pub(crate) fn spawn_outage_recovery_from_runtime() { + let Some(context) = node_runtime_context() else { + return; + }; + spawn_outage_recovery(context.db, context.wallet, context.map); +} + +pub(crate) async fn retire_peer_requests_from_runtime(peer: &str) { + let Some(context) = node_runtime_context() else { + return; + }; + retire_peer_entries(context.map, peer).await; +} + async fn reconnect_replacement_inner(excluded_ip: &str) { // When an outgoing peer disappears, try to replace it with another // active node that is not already connected and is not the failed IP. @@ -185,6 +201,13 @@ async fn retry_dropped_outgoing(ip: String, port: u16) { return; }; + // 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 { + spawn_outage_recovery(context.db, context.wallet, context.map); + return; + } + let addr_string = format!("{ip}:{port}"); for attempt in 1..=3 { // A connection-manager drop is definitive, so the first reconnect is @@ -361,9 +384,8 @@ impl Connection { if ClientType::from_bytes(&connection_info.client_type) == Some(ClientType::Miner) && connection_info.ready { - spawn_monitor_update( + spawn_monitor_removal( ip.clone(), - MONITOR_ACTION_REMOVE, connection_info.wallet_short_address.clone(), port, ); @@ -507,9 +529,13 @@ impl Connection { } } drop(guard); + retire_peer_entries(command_map.clone(), &format!("{ip}:{port}")).await; if connection_type == ConnectionType::Outgoing { spawn_retry_dropped_outgoing(ip.clone(), port); } + if miner_connection_count().await == 0 { + spawn_outage_recovery_from_runtime(); + } break; } } @@ -590,26 +616,20 @@ impl Connection { || !info.local_setup_complete || !info.remote_setup_complete || !info.local_setup_acknowledged + || info.monitor_activation_pending { return false; } - info.ready = true; - info.catch_up_target = None; - spawn_monitor_update( + info.monitor_activation_pending = true; + spawn_monitor_activation( ip.clone(), - MONITOR_ACTION_ADD, info.wallet_short_address.clone(), port, - ); - Connection::client_checkup( Arc::clone(&info.stream), - ConnectionType::from_bytes(&info.connection_type).unwrap_or(ConnectionType::Incoming), - ip, - port, command_map, ); - true + false } pub fn mark_local_setup_complete( @@ -741,9 +761,7 @@ impl Connection { .collect() } - pub fn get_mapping_relay_peer_streams_with_keys( - &self, - ) -> Vec<(String, Arc>)> { + pub fn get_mapping_relay_peer_streams_with_keys(&self) -> Vec<(String, Arc>)> { self.connection_map .iter() .filter_map(|(key, connection_info)| { @@ -948,25 +966,55 @@ impl Connection { } } -fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port: u16) { +fn spawn_monitor_activation( + ip: String, + monitored_address: String, + port: u16, + stream: Arc>, + command_map: Arc>, +) { + let Some(context) = node_runtime_context() else { + warn!("[network_map] node runtime context is not initialized"); + tokio::spawn(async move { + remove_stream_from_memory(&stream).await; + }); + return; + }; + if !Wallet::short_address_validation(&monitored_address) { + return; + } + let monitoring_address = context.wallet.saved.short_address.clone(); + if monitored_address == monitoring_address { + return; + } + tokio::spawn(async move { - let Some(context) = node_runtime_context() else { - warn!("[network_map] node runtime context is not initialized"); - return; - }; - if !Wallet::short_address_validation(&monitored_address) { - return; - } - let monitoring_address = context.wallet.saved.short_address.clone(); - if monitored_address == monitoring_address { + // 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 { + sleep(Duration::from_millis(100)).await; + } + }) + .await; + if local_activation.is_err() { + warn!( + "[network_map] connection setup timed out waiting for local membership activation: address={} peer={}:{}", + monitoring_address, ip, port + ); + remove_stream_from_memory(&stream).await; return; } + + // Generate ordering fields only after activation. A timestamp created + // before the wait could become stale relative to the activating event. let timestamp = crate::Utc::now().timestamp_millis() as u64; let modified_block = crate::records::block_height::get_block_height::get_height(&context.db) .saturating_add(1); let signature = NodeInfo::monitor_signature( - action, + MONITOR_ACTION_ADD, &monitored_address, &monitoring_address, &ip, @@ -976,7 +1024,7 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port: ) .await; let edit = SignedMonitorEdit { - action, + action: MONITOR_ACTION_ADD, monitored_address: monitored_address.clone(), monitoring_address, target_ip: ip.clone(), @@ -992,33 +1040,97 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port: wallet: context.wallet.clone(), connections_key: format!("{ip}:{port}"), }; - if action == MONITOR_ACTION_ADD { - for attempt in 1..=40 { - let RpcResponse::Binary(response) = NodeInfo::add_monitor(params.clone()).await; - if response.as_slice() == b"Success" { - return; - } - if attempt == 40 { - warn!( - "[network_map] monitor-add failed after {attempt} attempts: monitored={} monitoring={} error={}", - monitored_address, - params.edit.monitoring_address, - String::from_utf8_lossy(&response) - ); - return; - } - sleep(Duration::from_millis(250)).await; - } - } else { - let RpcResponse::Binary(response) = NodeInfo::remove_monitor(params.clone()).await; - if response.as_slice() != b"Success" { - warn!( - "[network_map] monitor-remove failed: monitored={} monitoring={} error={}", - monitored_address, - params.edit.monitoring_address, - String::from_utf8_lossy(&response) - ); - } + let RpcResponse::Binary(response) = NodeInfo::add_monitor(params.clone()).await; + if response.as_slice() != b"Success" { + warn!( + "[network_map] connection setup monitor-add failed: monitored={} monitoring={} error={}", + monitored_address, + params.edit.monitoring_address, + String::from_utf8_lossy(&response) + ); + remove_stream_from_memory(&stream).await; + return; + } + + let connection_type = { + let ip_bytes = ip_to_binary(&ip); + let mut connections = CONNECTIONS.write().await; + let Some(connection) = connections.as_mut() else { + return; + }; + let Some((key, info)) = connection.connection_map.iter_mut().find(|(key, info)| { + key.ip == ip_bytes + && key.port == port + && Arc::ptr_eq(&info.stream, &stream) + && info.monitor_activation_pending + && !info.ready + }) else { + return; + }; + + info.monitor_activation_pending = false; + info.ready = true; + info.catch_up_target = None; + ConnectionType::from_bytes(&key.connection_type).unwrap_or(ConnectionType::Incoming) + }; + + Connection::client_checkup(stream, connection_type, ip, port, command_map); + }); +} + +fn spawn_monitor_removal(ip: String, monitored_address: String, port: u16) { + let Some(context) = node_runtime_context() else { + warn!("[network_map] node runtime context is not initialized"); + return; + }; + if !Wallet::short_address_validation(&monitored_address) { + return; + } + let monitoring_address = context.wallet.saved.short_address.clone(); + if monitored_address == monitoring_address { + return; + } + + let timestamp = crate::Utc::now().timestamp_millis() as u64; + let modified_block = + crate::records::block_height::get_block_height::get_height(&context.db).saturating_add(1); + + tokio::spawn(async move { + let signature = NodeInfo::monitor_signature( + MONITOR_ACTION_REMOVE, + &monitored_address, + &monitoring_address, + &ip, + timestamp, + modified_block, + &context.wallet, + ) + .await; + let edit = SignedMonitorEdit { + action: MONITOR_ACTION_REMOVE, + monitored_address: monitored_address.clone(), + monitoring_address, + target_ip: ip, + modified_timestamp: timestamp, + modified_block, + modified_signature: signature, + }; + let params = MonitorAddressParams { + map: context.map.clone(), + edit, + remote_ip: String::new(), + db: context.db.clone(), + wallet: context.wallet.clone(), + connections_key: format!("disconnect:{port}"), + }; + let RpcResponse::Binary(response) = NodeInfo::remove_monitor(params.clone()).await; + if response.as_slice() != b"Success" { + warn!( + "[network_map] monitor-remove failed: monitored={} monitoring={} error={}", + monitored_address, + params.edit.monitoring_address, + String::from_utf8_lossy(&response) + ); } }); } diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index b69d70b..6096604 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -130,6 +130,24 @@ impl NodeInfo { } } + fn revive_deleted_membership(node: &mut NodeInfo, edit: &SignedNodeEdit) -> bool { + let accepted_ip_change = node.ip != edit.ip; + node.ip = edit.ip.clone(); + node.port = edit.port; + node.added_by = edit.modified_by.clone(); + node.added_timestamp = edit.modified_timestamp; + node.added_signature = edit.modified_signature.clone(); + + // Keep the new lifecycle inactive until a newer signed monitor-add + // proves that connection setup reached the operational state. + node.deleted_timestamp = edit.modified_timestamp; + // Zero is the deterministic pending marker: no historical block has + // yet activated this lifecycle, and every peer derives the same map. + node.deleted_block = 0; + node.monitoring.clear(); + accepted_ip_change + } + fn mark_conflict_loser( address_map: &mut HashMap, loser_address: &str, @@ -497,12 +515,7 @@ impl NodeInfo { existing_node.added_timestamp <= existing_node.deleted_timestamp; let mut accepted_ip_change = false; if current_is_old_cycle || Self::membership_is_preferred(&edit, existing_node) { - accepted_ip_change = existing_node.ip != edit.ip; - existing_node.ip = edit.ip.clone(); - existing_node.port = edit.port; - existing_node.added_by = edit.modified_by; - existing_node.added_timestamp = edit.modified_timestamp; - existing_node.added_signature = edit.modified_signature; + accepted_ip_change = Self::revive_deleted_membership(existing_node, &edit); } if accepted_ip_change { let mut monitor_state = MONITOR_EVENT_STATE.lock().await; @@ -989,12 +1002,8 @@ impl NodeInfo { if current_is_old_cycle || Self::membership_is_preferred(&edit, existing_node) { - let accepted_ip_change = existing_node.ip != edit.ip; - existing_node.ip = edit.ip.clone(); - existing_node.port = edit.port; - existing_node.added_by = edit.modified_by.clone(); - existing_node.added_timestamp = edit.modified_timestamp; - existing_node.added_signature = edit.modified_signature.clone(); + let accepted_ip_change = + Self::revive_deleted_membership(existing_node, &edit); if accepted_ip_change { let mut monitor_state = MONITOR_EVENT_STATE.lock().await; Self::discard_old_target_ip_events_from( @@ -1146,6 +1155,34 @@ mod tests { assert!(NodeInfo::validate_snapshot_structure(&map).is_err()); } + #[test] + fn reconnect_revives_membership_before_snapshot_is_exposed() { + let mut node = snapshot_node("1.2.3.5", 10, &["first.cltc"]); + node.blocks_mined = 95; + node.deleted_timestamp = 20; + node.deleted_block = 100; + + let edit = SignedNodeEdit { + address: "node.cltc".to_string(), + ip: "1.2.3.5".to_string(), + port: 50051, + modified_by: "new-sponsor.cltc".to_string(), + modified_timestamp: 30, + modified_signature: "11".repeat(Wallet::SIGNATURE_LENGTH), + }; + + assert!(!NodeInfo::revive_deleted_membership(&mut node, &edit)); + assert_eq!(node.added_timestamp, 30); + assert_eq!(node.deleted_timestamp, 30); + assert_eq!(node.deleted_block, 0); + assert!(node.monitoring.is_empty()); + assert_eq!(node.blocks_mined, 95); + + let mut map = HashMap::new(); + map.insert("node.cltc".to_string(), node); + assert!(NodeInfo::validate_snapshot_structure(&map).is_ok()); + } + #[test] fn complete_snapshot_rejects_deleted_or_duplicate_active_monitors() { let mut map = HashMap::new(); diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index c1f8d08..07bb1e4 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -82,6 +82,30 @@ mod tests { assert!(NodeInfo::monitor_event_order(&remove) > NodeInfo::monitor_event_order(&add)); } + #[test] + fn only_a_newer_monitor_add_activates_pending_membership() { + let mut pending = node(&[]); + pending.added_timestamp = 300; + pending.deleted_timestamp = 300; + + let mut edit = SignedMonitorEdit { + action: MONITOR_ACTION_ADD, + monitored_address: "target".to_string(), + monitoring_address: "monitor".to_string(), + target_ip: "1.2.3.4".to_string(), + modified_timestamp: 300, + modified_block: 50, + modified_signature: "signature".to_string(), + }; + assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending)); + + edit.modified_timestamp = 301; + assert!(NodeInfo::can_activate_pending_membership(&edit, &pending)); + + edit.action = MONITOR_ACTION_REMOVE; + assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending)); + } + #[test] fn signed_monitor_wire_record_round_trips() { let monitored = "ab13318c26250b048db92920a80a86127c933b0c.cltc".to_string(); @@ -259,14 +283,17 @@ impl NodeInfo { ) } + fn can_activate_pending_membership(edit: &SignedMonitorEdit, node: &NodeInfo) -> bool { + edit.action == MONITOR_ACTION_ADD && edit.modified_timestamp > node.deleted_timestamp + } + pub(super) fn discard_old_target_ip_events_from( state: &mut HashMap, address: &str, current_ip: &str, ) { - state.retain(|_, event| { - event.monitored_address != address || event.target_ip == current_ip - }); + state + .retain(|_, event| event.monitored_address != address || event.target_ip == current_ip); } async fn broadcast_monitor_event( @@ -356,6 +383,7 @@ impl NodeInfo { } if monitored.deleted_timestamp > 0 && monitored.added_timestamp <= monitored.deleted_timestamp + && !Self::can_activate_pending_membership(edit, monitored) { return Err( "monitor target must have a newer membership record before reconnection" @@ -449,7 +477,10 @@ impl NodeInfo { } let local_short = wallet.saved.short_address.clone(); - if remote_ip.is_empty() && edit.monitoring_address == local_short { + if remote_ip.is_empty() + && edit.monitoring_address == local_short + && edit.modified_signature.is_empty() + { edit.modified_timestamp = Utc::now().timestamp_millis() as u64; edit.modified_block = get_height(&db).saturating_add(1); edit.modified_signature = Self::monitor_signature( diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 2137b77..9af66ff 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -13,6 +13,11 @@ use crate::sled::Db; use std::collections::HashSet; impl NodeInfo { + 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) + } + pub async fn has_reciprocal_sponsorship( local_address: &str, operational_peer_wallets: &[String], @@ -33,8 +38,16 @@ impl NodeInfo { .unwrap_or(false) } - pub(crate) async fn signed_mapping_state( - ) -> (Vec, Vec) { + pub(crate) async fn is_active_address(address: &str) -> bool { + ADDRESS_MAP + .lock() + .await + .get(address) + .map(|node| node.deleted_timestamp == 0) + .unwrap_or(false) + } + + pub(crate) async fn signed_mapping_state() -> (Vec, Vec) { let map = ADDRESS_MAP.lock().await; let monitor_events = MONITOR_EVENT_STATE.lock().await; @@ -282,8 +295,9 @@ impl NodeInfo { let map = ADDRESS_MAP.lock().await; if let Some(node_info) = map.get(address) { // Deleted nodes remain valid for blocks before their recorded - // deletion height, which keeps historical validation deterministic. - return node_info.deleted_block == 0 || block_number < node_info.deleted_block; + // deletion height. A deleted record with block zero is a pending + // or rejected lifecycle and is never eligible to mine. + return Self::eligible_at_block(node_info, block_number); } false } @@ -448,6 +462,26 @@ mod tests { NODE_RECORD_FIXED_BYTES, }; + #[test] + fn pending_membership_is_not_mining_eligible() { + let mut pending = NodeInfo::new( + "198.51.100.10".to_string(), + 50050, + 0, + "sponsor.cltc".to_string(), + 300, + "signature".to_string(), + ); + pending.deleted_timestamp = 300; + pending.deleted_block = 0; + + assert!(!NodeInfo::eligible_at_block(&pending, 0)); + assert!(!NodeInfo::eligible_at_block(&pending, 10_000)); + + pending.deleted_timestamp = 0; + assert!(NodeInfo::eligible_at_block(&pending, 10_000)); + } + #[tokio::test] async fn bootstrap_snapshot_keeps_deletion_and_monitor_state() { let address = "1111111111111111111111111111111111111111.cltc"; diff --git a/src/records/memory/response_channels.rs b/src/records/memory/response_channels.rs index d95cf21..27c1be2 100644 --- a/src/records/memory/response_channels.rs +++ b/src/records/memory/response_channels.rs @@ -197,6 +197,24 @@ pub async fn delete_entry(map: Arc>, key: Byte3) { } } +pub async fn retire_peer_entries(map: Arc>, peer: &str) { + 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() { + continue; + } + + // Close the original waiter immediately while retaining the UID long + // enough to reject a late reply from the disconnected socket. + let (retired_tx, retired_rx) = mpsc::channel(1); + drop(retired_rx); + channel_pair.tx = retired_tx; + channel_pair.expires_at = Some(expires_at); + } +} + #[cfg(test)] mod tests { use super::*; @@ -212,4 +230,29 @@ mod tests { assert!(is_retired_entry(map, uid).await); } + + #[tokio::test] + async fn retiring_peer_entries_wakes_only_that_peers_waiters() { + let map = Arc::new(Mutex::new(Command::new())); + let (dead_uid, dead_tx, dead_rx) = reserve_entry_with_context( + map.clone(), + Some(2), + Some("203.0.113.10:50050".to_string()), + ) + .await; + let (live_uid, _live_tx, live_rx) = reserve_entry_with_context( + map.clone(), + Some(2), + Some("203.0.113.11:50050".to_string()), + ) + .await; + drop(dead_tx); + + retire_peer_entries(map.clone(), "203.0.113.10:50050").await; + + assert!(dead_rx.lock().await.recv().await.is_none()); + assert!(is_retired_entry(map.clone(), dead_uid).await); + assert!(!is_retired_entry(map.clone(), live_uid).await); + assert!(live_rx.lock().await.try_recv().is_err()); + } } diff --git a/src/records/memory/structs.rs b/src/records/memory/structs.rs index ea1c2e8..ffff99a 100644 --- a/src/records/memory/structs.rs +++ b/src/records/memory/structs.rs @@ -29,6 +29,7 @@ pub struct ConnectionInfo { pub local_setup_complete: bool, pub remote_setup_complete: bool, pub local_setup_acknowledged: bool, + pub monitor_activation_pending: bool, // Incoming peers that connect behind the tip retain the height they // originally need to reach. This avoids chasing a moving local tip. pub catch_up_target: Option, @@ -231,6 +232,7 @@ impl ConnectionInfo { local_setup_complete: false, remote_setup_complete: false, local_setup_acknowledged: false, + monitor_activation_pending: false, catch_up_target: None, ready: false, health: ConnectionHealth::Connected, diff --git a/src/rpc/server/connection_memory_manager.rs b/src/rpc/server/connection_memory_manager.rs index 3fb76e6..29de541 100644 --- a/src/rpc/server/connection_memory_manager.rs +++ b/src/rpc/server/connection_memory_manager.rs @@ -1,11 +1,12 @@ use crate::common::binary_conversions::{binary_to_ip, ip_to_binary}; use crate::log::warn; use crate::records::memory::connections::{ - set_stream_health, spawn_retry_dropped_outgoing, CONNECTIONS, + miner_connection_count, retire_peer_requests_from_runtime, set_stream_health, + spawn_outage_recovery_from_runtime, spawn_retry_dropped_outgoing, CONNECTIONS, }; use crate::records::memory::enums::{ClientType, ConnectionType}; use crate::records::memory::response_channels::{ - delete_entry, reserve_entry_with_context, Command, + delete_entry, reserve_entry_with_context, retire_peer_entries, Command, }; use crate::records::memory::structs::{ConnectionHealth, StoreConnectionParams}; use crate::rpc::command_maps::RPC_BLOCK_HEIGHT; @@ -127,8 +128,12 @@ async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) let duplicate_key = format!("{duplicate_ip}:{duplicate_port}"); match recorded_health { ConnectionHealth::Connected | ConnectionHealth::Busy => { - let health = - miner_stream_health(&duplicate_key, duplicate_stream.clone(), command_map).await; + let health = miner_stream_health( + &duplicate_key, + duplicate_stream.clone(), + command_map.clone(), + ) + .await; if matches!(health, ConnectionHealth::Connected | ConnectionHealth::Busy) { return false; } @@ -161,6 +166,8 @@ async fn purge_stale_duplicate_miner(ip: &str, command_map: Arc>) "[connection_manager] removing stale duplicate miner stream before accepting reconnect: peer={duplicate_key}" ); connection.drop_connection(connection_type, duplicate_ip, duplicate_port); + drop(guard); + retire_peer_entries(command_map, &duplicate_key).await; true } @@ -329,13 +336,18 @@ pub async fn remove_stream_from_memory(stream: &Arc>) { if let Some((connection_type, ip_bytes, port)) = matching_connection { let ip = crate::common::binary_conversions::binary_to_ip(ip_bytes); let dropped = connection.drop_connection(connection_type, ip.clone(), port); + drop(connection_instance); + retire_peer_requests_from_runtime(&format!("{ip}:{port}")).await; if connection_type == ConnectionType::Outgoing && dropped .as_ref() .and_then(|connection_info| ClientType::from_bytes(&connection_info.client_type)) == Some(ClientType::Miner) { - spawn_retry_dropped_outgoing(ip, port); + spawn_retry_dropped_outgoing(ip.clone(), port); + } + if miner_connection_count().await == 0 { + spawn_outage_recovery_from_runtime(); } } } diff --git a/src/startup/connections.rs b/src/startup/connections.rs index 1dcf12b..4bfe518 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::{ - operational_peer_wallets, peer_connection_count, ready_outgoing_connection_count, + miner_connection_count, operational_peer_wallets, ready_outgoing_connection_count, refill_outgoing_connections_once, }; use crate::records::memory::network_mapping::NodeInfo; @@ -15,6 +15,24 @@ use crate::wallets::structures::Wallet; use crate::Arc; use crate::Duration; use crate::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; + +static OUTAGE_RECOVERY_ACTIVE: AtomicBool = AtomicBool::new(false); + +struct OutageRecoveryGuard; + +impl Drop for OutageRecoveryGuard { + fn drop(&mut self) { + OUTAGE_RECOVERY_ACTIVE.store(false, Ordering::SeqCst); + } +} + +fn try_start_outage_recovery() -> Option { + OUTAGE_RECOVERY_ACTIVE + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .ok() + .map(|_| OutageRecoveryGuard) +} pub async fn handle_connections( db: Db, @@ -148,6 +166,29 @@ async fn attempt_bootstrap_connections( Ok(false) } +pub fn spawn_outage_recovery(db: Db, wallet: Arc, map: Arc>) { + let Some(recovery_guard) = try_start_outage_recovery() else { + return; + }; + + tokio::spawn(async move { + let _recovery_guard = recovery_guard; + + // A miner socket still completing setup owns recovery even though it + // is not ready for normal traffic yet. + if miner_connection_count().await != 0 { + return; + } + + info!("[reconnect] all miner peers are offline; starting coordinated outage recovery"); + match attempt_bootstrap_connections(db, wallet, map, "reconnect").await { + Ok(true) => {} + Ok(false) => warn!("[reconnect] coordinated outage recovery found no available peer"), + Err(err) => warn!("[reconnect] coordinated outage recovery aborted: {err}"), + } + }); +} + pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc, map: Arc>) { let outgoing_connections = crate::Settings::load() .map(|settings| settings.outgoing_connections) @@ -164,20 +205,8 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc, map: Arc {} - Ok(false) => {} - Err(err) => warn!("[reconnect] bootstrap recovery aborted: {err}"), - } + if miner_connection_count().await == 0 { + spawn_outage_recovery(db.clone(), wallet.clone(), map.clone()); continue; } @@ -190,3 +219,17 @@ pub fn spawn_isolated_bootstrap_recovery(db: Db, wallet: Arc, map: Arc