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

network mapping reconnection fixes

This commit is contained in:
contractless 2026-08-16 15:17:18 -06:00
parent d2a13cb134
commit a370c18825
9 changed files with 584 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
src/bin/lookup_online_node_count.rs Normal file
View File

@ -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<usize, String> {
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<String> = 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<u8> {
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<u8>]) -> Vec<u8> {
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());
}
}

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

@ -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<NodeRuntimeContext> {
.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<Mutex<TcpStream>>)> {
pub fn get_mapping_relay_peer_streams_with_keys(&self) -> Vec<(String, Arc<Mutex<TcpStream>>)> {
self.connection_map
.iter()
.filter_map(|(key, connection_info)| {
@ -948,10 +966,18 @@ impl Connection {
}
}
fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port: u16) {
tokio::spawn(async move {
fn spawn_monitor_activation(
ip: String,
monitored_address: String,
port: u16,
stream: Arc<Mutex<TcpStream>>,
command_map: Arc<Mutex<Command>>,
) {
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) {
@ -961,12 +987,34 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port:
if monitored_address == monitoring_address {
return;
}
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.
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,24 +1040,89 @@ 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 {
if response.as_slice() != b"Success" {
warn!(
"[network_map] monitor-add failed after {attempt} attempts: monitored={} monitoring={} error={}",
"[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;
}
sleep(Duration::from_millis(250)).await;
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;
}
} else {
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!(
@ -1019,7 +1132,6 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port:
String::from_utf8_lossy(&response)
);
}
}
});
}

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

@ -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<String, NodeInfo>,
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();

39
src/records/memory/network_mapping/monitor.rs
View File

@ -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<String, SignedMonitorEdit>,
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(

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

@ -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<SyncedNodeState>, Vec<SignedMonitorEdit>) {
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<SyncedNodeState>, Vec<SignedMonitorEdit>) {
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";

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

@ -197,6 +197,24 @@ pub async fn delete_entry(map: Arc<Mutex<Command>>, key: Byte3) {
}
}
pub async fn retire_peer_entries(map: Arc<Mutex<Command>>, 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());
}
}

2
src/records/memory/structs.rs
View File

@ -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<u32>,
@ -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,

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

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

73
src/startup/connections.rs
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::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<OutageRecoveryGuard> {
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<Wallet>, map: Arc<Mutex<Command>>) {
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<Wallet>, map: Arc<Mutex<Command>>) {
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<Wallet>, map: Arc<M
continue;
}
if peer_connection_count().await == 0 {
info!("[reconnect] no operational peers remain; retrying bootstrap recovery");
match attempt_bootstrap_connections(
db.clone(),
wallet.clone(),
map.clone(),
"reconnect",
)
.await
{
Ok(true) => {}
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<Wallet>, map: Arc<M
}
});
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn outage_recovery_has_one_owner() {
OUTAGE_RECOVERY_ACTIVE.store(false, Ordering::SeqCst);
let owner = try_start_outage_recovery().expect("first recovery should acquire ownership");
assert!(try_start_outage_recovery().is_none());
drop(owner);
assert!(try_start_outage_recovery().is_some());
}
}