bootstrap fixes

Cargo.lock

@@ -661,9 +661,9 @@ checksum = "a246d82be1c9d791c5dfde9a2bd045fc3cbba3fa2b11ad558f27d01712f00569"
 
 [[package]]
 name = "encrypted_images"
-version = "1.4.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "761fe0fc7059c1e9cb4217f49263b146d269322354aeefb8fc519a60b2e8cef1"
+checksum = "dda784ca648e472fc1bb0e408cb60b22e6b1647511d3c8f7144dadcf7133f605"
 dependencies = [
  "base64 0.21.7",
  "encoding",

Cargo.toml

@@ -26,7 +26,7 @@ codegen-units = 1
 
 [dependencies]
 rpassword = "7.1.0"
-encrypted_images = "1.4.0"
+encrypted_images = "1.4.1"
 skein = "0.1.0"
 chrono = "0.4"
 serde = { version = "1.0.108", features = ["derive"] }

src/orphans/deep_sync_rollback.rs

@@ -28,6 +28,13 @@ pub async fn deep_sync_rollback(mut params: CheckUp, wallet: Arc<Wallet>) {
                             // A hash mismatch this far behind means the local
                             // chain must step back until the torrents agree.
                             if local_torrent.info.info_hash != remote_torrent.info.info_hash {
+                                if params.local_height == 0 {
+                                    warn!(
+                                        "[orphan][deep_sync_rollback] remote genesis differs from local genesis; refusing to roll back below genesis"
+                                    );
+                                    break;
+                                }
+
                                 if !params.node_syncing {
                                     begin_reorg_lock().await;
                                 }
@@ -45,7 +52,7 @@ pub async fn deep_sync_rollback(mut params: CheckUp, wallet: Arc<Wallet>) {
                                 undo_transactions(undo_transactions_params, wallet.clone())
                                     .await
                                     .ok();
-                                params.local_height -= 1;
+                                params.local_height = params.local_height.saturating_sub(1);
                             } else {
                                 // Matching torrents mark the shared ancestor.
                                 break;

src/orphans/orphan_checkup.rs

@@ -9,7 +9,6 @@ use crate::records::memory::torrent_status::{
 };
 use crate::records::unpack_block::load_by_binary_data::load_block_from_binary;
 use crate::records::unpack_block::unpack_header::load_block_header;
-use crate::rpc::client::block_hash_vote::request_block_hash_at_height;
 use crate::torrent::structs::{DownloadSave, Torrent};
 use crate::torrent::torrenting_system::create_file::combine_pieces;
 use crate::torrent::torrenting_system::download_locks::acquire_candidate_download;
@@ -106,18 +105,6 @@ async fn candidate_attaches_before_rollback(
     // Metadata may choose a candidate, but only downloaded block bytes can
     // prove the rollback is safe.
     torrent.verify(height, &params.db, wallet).await?;
-    let peer_canonical_hash = request_block_hash_at_height(
-        params.stream.clone(),
-        params.map.clone(),
-        params.connections_key.clone(),
-        height,
-    )
-    .await?;
-    if peer_canonical_hash != torrent.info.block_hash {
-        return Err(format!(
-            "Staged candidate is not peer canonical at height {height}."
-        ));
-    }
 
     let _download_guard = acquire_candidate_download(height, &torrent.info.info_hash, true).await?;
 

src/records/memory/connections.rs

@@ -823,13 +823,6 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port:
         } else {
             NodeInfo::remove_monitor(params).await
         };
-        NodeInfo::broadcast_address_state(
-            context.map.clone(),
-            &monitored_address,
-            "",
-            &format!("{ip}:{port}"),
-        )
-        .await;
     });
 }
 

src/records/memory/network_mapping/monitor.rs

@@ -1,9 +1,15 @@
 use super::*;
+use crate::records::memory::response_channels::reserve_transient_entry_with_context;
 use crate::records::wallet_registry::resolve_pubkey_from_short_address;
+use crate::rpc::command_maps::{RPC_NETWORK_MONITOR_ADD, RPC_NETWORK_MONITOR_REMOVE};
 
 pub const MONITOR_ACTION_ADD: u8 = 1;
 pub const MONITOR_ACTION_REMOVE: u8 = 2;
 
+lazy_static! {
+    static ref MONITOR_EVENTS_SEEN: Mutex<HashMap<String, u64>> = Mutex::new(HashMap::new());
+}
+
 impl NodeInfo {
     pub async fn monitor_signature(
         action: u8,
@@ -99,12 +105,91 @@ impl NodeInfo {
         Self::apply_monitor(params, MONITOR_ACTION_REMOVE).await
     }
 
+    fn monitor_event_key(edit: &SignedMonitorEdit) -> String {
+        format!(
+            "{}:{}:{}:{}:{}:{}",
+            edit.action,
+            edit.monitored_address,
+            edit.monitoring_address,
+            edit.target_ip,
+            edit.modified_timestamp,
+            edit.modified_signature
+        )
+    }
+
+    async fn remember_monitor_event(edit: &SignedMonitorEdit) -> bool {
+        let key = Self::monitor_event_key(edit);
+        let mut seen = MONITOR_EVENTS_SEEN.lock().await;
+        let now = Utc::now().timestamp_millis() as u64;
+        seen.retain(|_, timestamp| now.saturating_sub(*timestamp) < 3_600_000);
+        seen.insert(key, now).is_none()
+    }
+
+    async fn broadcast_monitor_event(
+        map: Arc<Mutex<Command>>,
+        edit: &SignedMonitorEdit,
+        source_ip: &str,
+    ) {
+        let message_type = match edit.action {
+            MONITOR_ACTION_ADD => RPC_NETWORK_MONITOR_ADD,
+            MONITOR_ACTION_REMOVE => RPC_NETWORK_MONITOR_REMOVE,
+            _ => return,
+        };
+        let monitored_bytes = match Wallet::short_address_to_bytes(&edit.monitored_address) {
+            Some(bytes) => bytes,
+            None => return,
+        };
+        let monitoring_bytes = match Wallet::short_address_to_bytes(&edit.monitoring_address) {
+            Some(bytes) => bytes,
+            None => return,
+        };
+        let target_ip_bytes = ip_to_binary(&edit.target_ip);
+        let timestamp_bytes = edit.modified_timestamp.to_le_bytes();
+        let signature_bytes = match decode(&edit.modified_signature) {
+            Ok(bytes) => bytes,
+            Err(_) => return,
+        };
+        let streams = {
+            let connections_lock = CONNECTIONS.read().await;
+            connections_lock
+                .as_ref()
+                .map(|connection| connection.get_all_ready_peer_streams_with_keys())
+                .unwrap_or_default()
+        };
+
+        for (peer_key, unlocked_stream) in streams {
+            let Some((peer_ip, _)) = peer_key.rsplit_once(':') else {
+                continue;
+            };
+            if !source_ip.is_empty() && peer_ip == source_ip {
+                continue;
+            }
+            let hashmap_key = reserve_transient_entry_with_context(
+                map.clone(),
+                Some(message_type),
+                Some(peer_key.clone()),
+            )
+            .await;
+            let mut message: Vec<u8> = Vec::new();
+            message.push(message_type);
+            message.extend_from_slice(&hashmap_key);
+            message.push(edit.action);
+            message.extend_from_slice(&monitored_bytes);
+            message.extend_from_slice(&monitoring_bytes);
+            message.extend_from_slice(&target_ip_bytes);
+            message.extend_from_slice(&timestamp_bytes);
+            message.extend_from_slice(&signature_bytes);
+            RpcResponse::send_raw(&unlocked_stream, Some(&peer_key), &message).await;
+        }
+    }
+
     async fn apply_monitor(params: MonitorAddressParams, action: u8) -> RpcResponse {
         let MonitorAddressParams {
             mut edit,
             remote_ip,
             db,
             wallet,
+            map,
             ..
         } = params;
         let current_timestamp = Utc::now().timestamp_millis() as u64;
@@ -131,6 +216,11 @@ impl NodeInfo {
             return RpcResponse::Binary(b"Error: Could not validate monitor signature".to_vec());
         }
 
+        let is_new_event = Self::remember_monitor_event(&edit).await;
+        if !is_new_event {
+            return RpcResponse::Binary(b"Success".to_vec());
+        }
+
         {
             let mut address_map = ADDRESS_MAP.lock().await;
             let Some(monitored) = address_map.get_mut(&edit.monitored_address) else {
@@ -169,6 +259,7 @@ impl NodeInfo {
         }
 
         Self::persist_recovery_snapshot("monitor update").await;
+        Self::broadcast_monitor_event(map, &edit, &remote_ip).await;
 
         RpcResponse::Binary(b"Success".to_vec())
     }

src/rpc/client/genesis_compat.rs

@@ -0,0 +1,46 @@
+use crate::common::check_genesis::genesis_checkup;
+use crate::records::memory::response_channels::Command;
+use crate::records::unpack_block::unpack_header::load_block_header;
+use crate::rpc::client::block_hash_vote::request_block_hash_at_height;
+use crate::{Arc, Mutex, TcpStream};
+
+async fn local_genesis_hash() -> Result<Option<String>, String> {
+    if !genesis_checkup().await {
+        return Ok(None);
+    }
+
+    let header = load_block_header(0).await?;
+    Ok(Some(header.hash().await))
+}
+
+pub async fn ensure_compatible_genesis(
+    stream: Arc<Mutex<TcpStream>>,
+    map: Arc<Mutex<Command>>,
+    connections_key: String,
+    remote_height: u32,
+) -> Result<(), String> {
+    let local_hash = local_genesis_hash().await?;
+    let remote_hash =
+        match request_block_hash_at_height(stream, map, connections_key.clone(), 0).await {
+            Ok(hash) => Some(hash),
+            Err(err) if err.contains("Block 0 not found") => None,
+            Err(err) => {
+                return Err(format!(
+                    "remote genesis compatibility check failed for {connections_key}: {err}"
+                ));
+            }
+        };
+
+    match (local_hash, remote_hash) {
+        (Some(local), Some(remote)) if local != remote => Err(format!(
+            "incompatible genesis from peer {connections_key}: local={local} remote={remote}"
+        )),
+        (Some(_), None) => Err(format!(
+            "peer {connections_key} has no genesis while local chain already has one"
+        )),
+        (None, None) if remote_height > 0 => Err(format!(
+            "peer {connections_key} reports height {remote_height} but has no genesis"
+        )),
+        _ => Ok(()),
+    }
+}

src/rpc/client/handshake_processing.rs

@@ -21,6 +21,7 @@ use crate::records::memory::connections::{
 use crate::records::memory::enums::{ClientType, ConnectionType};
 use crate::records::memory::response_channels::{reserve_entry, Command};
 use crate::records::memory::structs::{Connection, StoreConnectionParams};
+use crate::rpc::client::genesis_compat::ensure_compatible_genesis;
 use crate::rpc::client::handshake::connect_and_handshake;
 use crate::rpc::client::register_wallet::register_connected_wallet;
 use crate::rpc::client::structs::{Connect, Handshake};
@@ -170,6 +171,13 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(),
         let local_height = get_height(&params.db);
         let remote_height =
             request_remote_height(stream.clone(), params.map.clone(), current_key.clone()).await?;
+        ensure_compatible_genesis(
+            stream.clone(),
+            params.map.clone(),
+            current_key.clone(),
+            remote_height,
+        )
+        .await?;
         let local_genesis_exists = genesis_checkup().await;
 
         if !local_genesis_exists || remote_height > local_height + 10 {
@@ -390,6 +398,20 @@ pub async fn process_handshake_response(
             "Remote height lookup failed before network join: {err}"
         ))
     })?;
+    ensure_compatible_genesis(
+        broadcast_stream.clone(),
+        params.map.clone(),
+        connections_key.clone(),
+        remote_height,
+    )
+    .await
+    .map_err(|err| {
+        let connections_key = connections_key.clone();
+        tokio::spawn(async move {
+            remove_key_from_memory(&connections_key).await;
+        });
+        io::Error::other(format!("Genesis compatibility check failed: {err}"))
+    })?;
     let self_add_block = crate::records::memory::network_mapping::SELF_ADD_BLOCK;
     let self_add_limit = crate::records::memory::network_mapping::NodeInfo::self_add_limit_height();
     let join_mode = if crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height(

src/rpc/client/mod.rs

@@ -1,5 +1,6 @@
 // The rpc client module contains the standalone client-side handshake and sync helpers.
 pub mod block_hash_vote;
+pub mod genesis_compat;
 pub mod handshake;
 pub mod handshake_message;
 pub mod handshake_processing;

src/rpc/command_maps.rs

@@ -60,6 +60,8 @@ pub const RPC_LATEST_ADDRESS_TRANSACTIONS: u8 = 46;
 pub const RPC_MARKETING_CAMPAIGN_HISTORY: u8 = 47;
 pub const RPC_TOKEN_CATALOG: u8 = 48;
 pub const RPC_COLLATERAL_STATUS: u8 = 49;
+pub const RPC_NETWORK_MONITOR_ADD: u8 = 50;
+pub const RPC_NETWORK_MONITOR_REMOVE: u8 = 51;
 pub const RPC_REPLY: u8 = 255;
 pub const MAX_RPC_REPLY_BYTES: usize = 64 * 1024 * 1024;
 

src/rpc/commands/mod.rs

@@ -19,6 +19,8 @@ pub mod latest_block;
 pub mod marketing_campaign_history;
 pub mod memory_by_signature;
 pub mod network_info;
+pub mod network_monitor_add;
+pub mod network_monitor_remove;
 pub mod nft_list;
 pub mod nft_lookup;
 pub mod random_node;

src/rpc/commands/network_monitor_add.rs

@@ -1,3 +1,4 @@
+use crate::records::memory::enums::ClientType;
 use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit};
 use crate::records::memory::network_mapping::NodeInfo;
 use crate::records::memory::response_channels::Command;
@@ -15,6 +16,7 @@ pub async fn network_monitor_add(
     db: &Db,
     wallet: Arc<Wallet>,
     map: Arc<Mutex<Command>>,
+    client_type: ClientType,
 ) -> Result<(u32, RpcResponse), String> {
     let (uid, _) =
         read_bytes_from_stream::read_uid_from_stream(connections_key, stream_locked.clone())
@@ -45,6 +47,13 @@ pub async fn network_monitor_add(
             .await?;
     let remote_ip = read_bytes_from_stream::read_caller_ip(stream_locked).await?;
 
+    if client_type != ClientType::Miner {
+        return Ok((
+            uid,
+            RpcResponse::Binary(b"Error: monitor updates are miner-peer only".to_vec()),
+        ));
+    }
+
     let result = NodeInfo::add_monitor(MonitorAddressParams {
         map,
         edit: SignedMonitorEdit {

src/rpc/commands/network_monitor_remove.rs

@@ -1,3 +1,4 @@
+use crate::records::memory::enums::ClientType;
 use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit};
 use crate::records::memory::network_mapping::NodeInfo;
 use crate::records::memory::response_channels::Command;
@@ -15,6 +16,7 @@ pub async fn network_monitor_remove(
     db: &Db,
     wallet: Arc<Wallet>,
     map: Arc<Mutex<Command>>,
+    client_type: ClientType,
 ) -> Result<(u32, RpcResponse), String> {
     let (uid, _) =
         read_bytes_from_stream::read_uid_from_stream(connections_key, stream_locked.clone())
@@ -45,6 +47,13 @@ pub async fn network_monitor_remove(
             .await?;
     let remote_ip = read_bytes_from_stream::read_caller_ip(stream_locked).await?;
 
+    if client_type != ClientType::Miner {
+        return Ok((
+            uid,
+            RpcResponse::Binary(b"Error: monitor updates are miner-peer only".to_vec()),
+        ));
+    }
+
     let result = NodeInfo::remove_monitor(MonitorAddressParams {
         map,
         edit: SignedMonitorEdit {

src/rpc/commands/receive_torrent.rs

@@ -305,9 +305,26 @@ pub async fn torrent_submission(
                         }
                         Err(e) => {
                             if is_stale_next_height_error(&e) {
-                                info!(
+                                let current_height = get_height(&db_clone);
-                                    "[broadcast] ignored stale downloaded block: height={height} err={e}"
+                                if height > 0 && within_orphan_window(current_height, height) {
-                                );
+                                    warn!(
+                                        "[broadcast] stale downloaded block is inside orphan window: height={height} local_height={current_height} err={e}"
+                                    );
+                                    trigger_orphan_check(
+                                        &e,
+                                        height,
+                                        stream_clone,
+                                        &db_clone,
+                                        wallet_clone,
+                                        map_clone,
+                                        connections_key_clone,
+                                    )
+                                    .await;
+                                } else {
+                                    info!(
+                                        "[broadcast] ignored stale downloaded block: height={height} err={e}"
+                                    );
+                                }
                                 return;
                             }
                             error!("[broadcast] setup_download error: height={height} err={e}");

src/rpc/server/handshake.rs

@@ -14,6 +14,7 @@ use crate::records::memory::connections::{
 use crate::records::memory::response_channels::generate_uid;
 use crate::records::memory::response_channels::Command;
 use crate::records::memory::structs::Connection;
+use crate::rpc::client::genesis_compat::ensure_compatible_genesis;
 use crate::rpc::client::register_wallet::register_connected_wallet;
 use crate::rpc::client::syncing::node_syncing;
 use crate::rpc::client::wallet_registry_sync::sync_wallet_registry_with_retries;
@@ -25,7 +26,9 @@ 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;
+use crate::startup::network_broadcast::{
+    announce_self_to_network, get_network_mapping_for_address,
+};
 use crate::startup::remote_height::request_remote_height;
 use crate::wallets::structures::Wallet;
 use crate::Arc;
@@ -63,6 +66,13 @@ async fn sync_incoming_peer_before_operational(
     let local_height = get_height(db);
     let remote_height =
         request_remote_height(stream.clone(), map.clone(), connections_key.to_string()).await?;
+    ensure_compatible_genesis(
+        stream.clone(),
+        map.clone(),
+        connections_key.to_string(),
+        remote_height,
+    )
+    .await?;
     let local_genesis_exists = genesis_checkup().await;
 
     if local_genesis_exists && remote_height < local_height {
@@ -222,6 +232,18 @@ async fn complete_incoming_miner_setup(
                 return;
             }
         };
+    if let Err(err) = ensure_compatible_genesis(
+        stream.clone(),
+        map.clone(),
+        connections_key.to_string(),
+        remote_height,
+    )
+    .await
+    {
+        error!("[startup] incoming peer genesis compatibility check failed: {err}");
+        remove_key_from_memory(connections_key).await;
+        return;
+    }
     let self_add_allowed =
         crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height(
             remote_height,
@@ -247,6 +269,20 @@ async fn complete_incoming_miner_setup(
             remove_key_from_memory(connections_key).await;
             return;
         }
+        if let Err(err) = get_network_mapping_for_address(
+            stream.clone(),
+            map.clone(),
+            db,
+            wallet.clone(),
+            connections_key,
+            &short_address,
+        )
+        .await
+        {
+            error!("[startup] incoming peer local network record import failed: {err}");
+            remove_key_from_memory(connections_key).await;
+            return;
+        }
     }
     mark_peer_network_map_synced(connections_key).await;
 

src/rpc/server/rpc_command_loop.rs

@@ -567,6 +567,36 @@ pub async fn start_loop(
                     .send(&stream_locked, Some(&connections_key), uid)
                     .await;
             }
+            50 => {
+                // signed monitor-add event from a miner peer
+                let (uid, result) = commands::network_monitor_add::network_monitor_add(
+                    &connections_key,
+                    stream_locked.clone(),
+                    &db,
+                    wallet.clone(),
+                    map.clone(),
+                    client_type,
+                )
+                .await?;
+                result
+                    .send(&stream_locked, Some(&connections_key), uid)
+                    .await;
+            }
+            51 => {
+                // signed monitor-remove event from a miner peer
+                let (uid, result) = commands::network_monitor_remove::network_monitor_remove(
+                    &connections_key,
+                    stream_locked.clone(),
+                    &db,
+                    wallet.clone(),
+                    map.clone(),
+                    client_type,
+                )
+                .await?;
+                result
+                    .send(&stream_locked, Some(&connections_key), uid)
+                    .await;
+            }
             30 => {
                 // request node list
                 let (uid, _) = read_bytes_from_stream::read_uid_from_stream(

src/startup/network_broadcast.rs

@@ -105,6 +105,34 @@ pub async fn get_network_mapping(
     db: &Db,
     _wallet: Arc<Wallet>,
     connections_key: &str,
+) -> Result<(), String> {
+    get_network_mapping_inner(unlocked_stream, command_map, db, connections_key, None).await
+}
+
+pub async fn get_network_mapping_for_address(
+    unlocked_stream: Arc<Mutex<TcpStream>>,
+    command_map: Arc<Mutex<Command>>,
+    db: &Db,
+    _wallet: Arc<Wallet>,
+    connections_key: &str,
+    only_address: &str,
+) -> Result<(), String> {
+    get_network_mapping_inner(
+        unlocked_stream,
+        command_map,
+        db,
+        connections_key,
+        Some(only_address),
+    )
+    .await
+}
+
+async fn get_network_mapping_inner(
+    unlocked_stream: Arc<Mutex<TcpStream>>,
+    command_map: Arc<Mutex<Command>>,
+    db: &Db,
+    connections_key: &str,
+    only_address: Option<&str>,
 ) -> Result<(), String> {
     // request the remote peer's serialized node list and
     // import each advertised add/delete record locally
@@ -177,6 +205,13 @@ pub async fn get_network_mapping(
             }
         }
 
+        if only_address
+            .map(|wanted_address| wanted_address != address)
+            .unwrap_or(false)
+        {
+            continue;
+        }
+
         // Import signed map records as remote state. This verifies the original
         // signer but does not require the newly connecting node to be eligible
         // to add nodes locally after the mature-network gate.