From 32664eb2f495260968b18b0f8478dc00152a52f8 Mon Sep 17 00:00:00 2001 From: viraladmin <00purple@gmail.com> Date: Fri, 3 Jul 2026 03:27:14 -0600 Subject: [PATCH] syncing fixes --- src/records/memory/connections.rs | 1 + src/records/memory/network_mapping/add.rs | 65 +++++++++++++++---- src/records/memory/network_mapping/monitor.rs | 3 +- src/rpc/client/handshake_processing.rs | 1 + src/rpc/commands/receive_torrent.rs | 35 ++++++---- src/rpc/server/handshake.rs | 15 +---- src/startup/network_broadcast.rs | 24 ++++--- 7 files changed, 94 insertions(+), 50 deletions(-) diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index e8b983b..93537db 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -295,6 +295,7 @@ impl Connection { let removed = self.connection_map.remove(&connection_key); if let Some(connection_info) = removed.as_ref() { if ClientType::from_bytes(&connection_info.client_type) == Some(ClientType::Miner) + && connection_info.ready && is_normal_mode() { spawn_monitor_update( diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 9180f9b..3eb4143 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -7,6 +7,24 @@ fn connection_key_ip(connections_key: &str) -> Option<&str> { connections_key.rsplit_once(':').map(|(ip, _)| ip) } +fn signature_is_empty(signature: &str) -> bool { + signature.is_empty() + || decode(signature) + .map(|bytes| bytes.iter().all(|byte| *byte == 0)) + .unwrap_or(false) +} + +fn merge_monitors(existing: &mut Vec, monitors: Vec) -> bool { + let mut changed = false; + for monitor in monitors { + if !existing.iter().any(|known| known == &monitor) { + existing.push(monitor); + changed = true; + } + } + changed +} + impl NodeInfo { pub async fn import_signed_mapping_address( db: &Db, @@ -42,12 +60,12 @@ impl NodeInfo { let mut address_map = ADDRESS_MAP.lock().await; if let Some(existing_node) = address_map.get_mut(&edit.address) { - existing_node.monitoring = monitors.clone(); if existing_node.deleted_timestamp > 0 { existing_node.ip = edit.ip; existing_node.added_by = edit.modified_by; existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature; + existing_node.monitoring = monitors; existing_node.deleted_timestamp = 0_u64; existing_node.deleted_block = 0_u32; drop(address_map); @@ -57,12 +75,12 @@ impl NodeInfo { if existing_node.ip != edit.ip { return Err("active node must be deleted before changing IP".to_string()); } - if edit.modified_timestamp > existing_node.added_timestamp { + if edit.modified_timestamp < existing_node.added_timestamp { existing_node.added_by = edit.modified_by; existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature; - existing_node.monitoring = monitors; } + merge_monitors(&mut existing_node.monitoring, monitors); drop(address_map); Self::persist_recovery_snapshot("import existing").await; return Ok(()); @@ -216,15 +234,34 @@ impl NodeInfo { return RpcResponse::Binary(b"Error: Invalid network address".to_vec()); } - let unsigned_self_announcement = edit.modified_by.is_empty() + let unsigned_add_request = edit.modified_by.is_empty() + && signature_is_empty(&edit.modified_signature) && (edit.ip == remote_ip || connection_key_ip(&connections_key) .map(|connection_ip| edit.ip == connection_ip) .unwrap_or(false)); - // Unsigned self-announcements are sponsored by the receiving node - // after the claimed IP is tied to the live peer connection. - if unsigned_self_announcement { + if edit.modified_by.is_empty() && !unsigned_add_request { + return RpcResponse::Binary(b"Error: Invalid unsigned node add request".to_vec()); + } + + if unsigned_add_request { + let address_map = ADDRESS_MAP.lock().await; + if let Some(existing_node) = address_map.get(&edit.address) { + if existing_node.deleted_timestamp == 0 { + if existing_node.ip == edit.ip { + return RpcResponse::Binary(b"Success".to_vec()); + } + return RpcResponse::Binary( + b"Error: Active node must be deleted before changing IP".to_vec(), + ); + } + } + drop(address_map); + + // Unsigned add requests are sponsored by the receiving node only + // when the address is missing or was previously deleted. Active + // records remain canonical and are not re-signed on reconnect. edit.modified_timestamp = current_timestamp; edit.modified_by = wallet.saved.short_address.clone(); edit.modified_signature = @@ -322,7 +359,6 @@ impl NodeInfo { // same IP are still rejected, but deleted records remain in place // for historical validation. if let Some(existing_node) = address_map.get_mut(&edit.address) { - existing_node.monitoring = monitors.clone(); if existing_node.deleted_timestamp > 0 { existing_node.ip = edit.ip.clone(); existing_node.added_by = edit.modified_by.clone(); @@ -337,11 +373,13 @@ impl NodeInfo { b"Error: Active node must be deleted before changing IP".to_vec(), ); } - if edit.modified_timestamp > existing_node.added_timestamp { + if edit.modified_timestamp < existing_node.added_timestamp { existing_node.added_by = edit.modified_by.clone(); existing_node.added_timestamp = edit.modified_timestamp; existing_node.added_signature = edit.modified_signature.clone(); - existing_node.monitoring = monitors.clone(); + state_changed = true; + } + if merge_monitors(&mut existing_node.monitoring, monitors.clone()) { state_changed = true; } } @@ -398,15 +436,14 @@ impl NodeInfo { if state_changed && !remote_ip.is_empty() { let broadcast_map = map.clone(); - let broadcast_edit = edit.clone(); + let broadcast_address = edit.address.clone(); let broadcast_remote_ip = remote_ip.clone(); let broadcast_connections_key = connections_key.clone(); tokio::spawn(async move { - Self::broadcast_node( + Self::broadcast_address_state( broadcast_map, - &broadcast_edit, + &broadcast_address, &broadcast_remote_ip, - NodeEditType::Add, &broadcast_connections_key, ) .await; diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index f169a02..9ead624 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -149,10 +149,11 @@ impl NodeInfo { monitored.deleted_block = 0; } MONITOR_ACTION_REMOVE => { + let before = monitored.monitoring.len(); monitored .monitoring .retain(|monitor| monitor != &edit.monitoring_address); - if monitored.monitoring.is_empty() { + if before != monitored.monitoring.len() && monitored.monitoring.is_empty() { let deleted_block = get_height(&db) + 1; Self::mark_deleted_and_cascade( &mut address_map, diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 03cde20..213b1c4 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -410,6 +410,7 @@ pub async fn process_handshake_response( ¶ms.db.clone(), params.wallet.clone(), &connections_key, + true, ) .await .map_err(|err| { diff --git a/src/rpc/commands/receive_torrent.rs b/src/rpc/commands/receive_torrent.rs index d752847..356a0ba 100644 --- a/src/rpc/commands/receive_torrent.rs +++ b/src/rpc/commands/receive_torrent.rs @@ -34,6 +34,7 @@ use crate::Arc; use crate::Mutex; const LIVE_SYNC_HEIGHT_GAP: u32 = 10; +const SYNC_NON_OPERATIONAL_STAGE_WINDOW: u32 = 64; const STALE_NEXT_HEIGHT_ERROR: &str = "Incoming block is no longer the next expected height."; fn is_stale_next_height_error(error: &str) -> bool { @@ -414,21 +415,31 @@ pub async fn receive_torrent( { let local_height = get_height(db); if !within_orphan_window(local_height, block_number) { + if is_syncing_mode() + && block_number > local_height + && block_number <= local_height.saturating_add(SYNC_NON_OPERATIONAL_STAGE_WINDOW) + { + warn!( + "[broadcast] staging near-tip torrent from non-operational peer during sync: peer={connections_key} local_height={local_height} height={block_number}" + ); + } else { + warn!( + "[broadcast] ignored torrent from non-operational peer outside orphan window: peer={connections_key} local_height={local_height} height={block_number}" + ); + return Ok(( + uid, + RpcResponse::Binary( + "Torrent ignored from non-operational peer." + .as_bytes() + .to_vec(), + ), + )); + } + } else { warn!( - "[broadcast] ignored torrent from non-operational peer outside orphan window: peer={connections_key} local_height={local_height} height={block_number}" + "[broadcast] accepting torrent from non-operational peer within orphan window: peer={connections_key} local_height={local_height} height={block_number}" ); - return Ok(( - uid, - RpcResponse::Binary( - "Torrent ignored from non-operational peer." - .as_bytes() - .to_vec(), - ), - )); } - warn!( - "[broadcast] accepting torrent from non-operational peer within orphan window: peer={connections_key} local_height={local_height} height={block_number}" - ); } let outcome = torrent_submission( diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index 8745e5e..dd54662 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -25,7 +25,7 @@ use crate::rpc::server::structs::{CombineAndSendDataParams, HandshakeTestParams} use crate::rpc::server::tests::{endpoint_port, is_port_open}; use crate::sled::Db; use crate::sleep; -use crate::startup::network_broadcast::{announce_self_to_network, get_network_mapping}; +use crate::startup::network_broadcast::announce_self_to_network; use crate::startup::remote_height::request_remote_height; use crate::wallets::structures::Wallet; use crate::Arc; @@ -239,6 +239,7 @@ async fn complete_incoming_miner_setup( db, wallet.clone(), connections_key, + false, ) .await { @@ -246,18 +247,6 @@ async fn complete_incoming_miner_setup( remove_key_from_memory(connections_key).await; return; } - } else if let Err(err) = get_network_mapping( - stream.clone(), - map.clone(), - db, - wallet.clone(), - connections_key, - ) - .await - { - error!("[startup] incoming peer network map import failed: {err}"); - remove_key_from_memory(connections_key).await; - return; } mark_peer_network_map_synced(connections_key).await; diff --git a/src/startup/network_broadcast.rs b/src/startup/network_broadcast.rs index 53786ec..435e5b7 100644 --- a/src/startup/network_broadcast.rs +++ b/src/startup/network_broadcast.rs @@ -26,9 +26,11 @@ pub async fn announce_self_to_network( db: &Db, wallet: Arc, connections_key: &str, + import_peer_map: bool, ) -> Result<(), String> { - // announce the local node to the connected peer, then - // request its current network mapping on success + // Announce the local node to the connected peer. Joining nodes can + // explicitly request the peer's map after acceptance; incoming/server + // handshakes must not import a potentially stale joining peer map. let message_type = RPC_ADD_NETWORK_NODE; let (ip, _, _) = get_ip_and_port().await; @@ -84,14 +86,16 @@ pub async fn announce_self_to_network( )); } - get_network_mapping( - unlocked_stream, - command_map.clone(), - db, - wallet.clone(), - connections_key, - ) - .await?; + if import_peer_map { + get_network_mapping( + unlocked_stream, + command_map.clone(), + db, + wallet.clone(), + connections_key, + ) + .await?; + } Ok(()) }