syncing fixes

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(

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<String>, monitors: Vec<String>) -> 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
+        if edit.modified_by.is_empty() && !unsigned_add_request {
-        // after the claimed IP is tied to the live peer connection.
+            return RpcResponse::Binary(b"Error: Invalid unsigned node add request".to_vec());
-        if unsigned_self_announcement {
+        }
+
+        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;

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,

src/rpc/client/handshake_processing.rs

@@ -410,6 +410,7 @@ pub async fn process_handshake_response(
         &params.db.clone(),
         params.wallet.clone(),
         &connections_key,
+        true,
     )
     .await
     .map_err(|err| {

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(

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;
 

src/startup/network_broadcast.rs

@@ -26,9 +26,11 @@ pub async fn announce_self_to_network(
     db: &Db,
     wallet: Arc<Wallet>,
     connections_key: &str,
+    import_peer_map: bool,
 ) -> Result<(), String> {
-    // announce the local node to the connected peer, then
+    // Announce the local node to the connected peer. Joining nodes can
-    // request its current network mapping on success
+    // 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(
+    if import_peer_map {
-        unlocked_stream,
+        get_network_mapping(
-        command_map.clone(),
+            unlocked_stream,
-        db,
+            command_map.clone(),
-        wallet.clone(),
+            db,
-        connections_key,
+            wallet.clone(),
-    )
+            connections_key,
-    .await?;
+        )
+        .await?;
+    }
     Ok(())
 }