startup fixes

src/records/memory/connections.rs

@@ -992,12 +992,12 @@ fn spawn_monitor_activation(
     }
 
     tokio::spawn(async move {
-        // A reconnecting membership remains pending until an active sponsor's
+        // During reciprocal startup both peers can hold freshly sponsored
-        // monitor-add arrives. Do not let that pending node author its own
+        // pending memberships. Permit either side to author its monitor-add so
-        // monitor event first; monitor validation correctly rejects it.
+        // the two activation events cannot deadlock waiting on one another.
         let mapping_prerequisites = timeout(Duration::from_secs(30), async {
-            while !NodeInfo::is_active_address(&monitoring_address).await
+            while !NodeInfo::contains_address(&monitored_address).await
-                || !NodeInfo::contains_address(&monitored_address).await
+                || !NodeInfo::is_active_or_sponsored_pending(&monitoring_address).await
             {
                 sleep(Duration::from_millis(100)).await;
             }

src/records/memory/network_mapping/add.rs

@@ -1,5 +1,7 @@
 use super::*;
-use crate::records::memory::connections::drop_miner_connections_for_wallet;
+use crate::records::memory::connections::{
+    drop_miner_connections_for_wallet, operational_peer_wallets,
+};
 use crate::records::memory::network_mapping::monitor::MONITOR_EVENT_STATE;
 use crate::records::memory::network_mapping::structs::SyncedNodeState;
 use crate::records::memory::response_channels::reserve_transient_entry_with_context;
@@ -19,6 +21,48 @@ fn signature_is_empty(signature: &str) -> bool {
 }
 
 impl NodeInfo {
+    fn eligible_sponsor_endpoint_from_map(
+        address_map: &HashMap<String, NodeInfo>,
+        operational_addresses: &HashSet<String>,
+        current_timestamp: u64,
+        excluded_address: &str,
+    ) -> Option<String> {
+        let mut candidates: Vec<(&String, &NodeInfo)> = address_map
+            .iter()
+            .filter(|(address, node)| {
+                address.as_str() != excluded_address
+                    && operational_addresses.contains(address.as_str())
+                    && node.deleted_timestamp == 0
+                    && current_timestamp.saturating_sub(node.added_timestamp) >= ONE_HOUR_MILLIS
+                    && node.blocks_mined >= 100
+            })
+            .collect();
+        candidates.sort_by(|(left_address, left), (right_address, right)| {
+            left.added_timestamp
+                .cmp(&right.added_timestamp)
+                .then_with(|| left_address.cmp(right_address))
+        });
+        candidates
+            .first()
+            .map(|(_, node)| format!("{}:{}", node.ip, node.port))
+    }
+
+    fn sponsor_redirect_suffix(
+        address_map: &HashMap<String, NodeInfo>,
+        operational_addresses: &HashSet<String>,
+        current_timestamp: u64,
+        excluded_address: &str,
+    ) -> String {
+        Self::eligible_sponsor_endpoint_from_map(
+            address_map,
+            operational_addresses,
+            current_timestamp,
+            excluded_address,
+        )
+        .map(|endpoint| format!(" sponsor={endpoint}"))
+        .unwrap_or_default()
+    }
+
     fn validate_snapshot_structure(replacement: &HashMap<String, NodeInfo>) -> Result<(), String> {
         let mut active_ips = HashMap::<String, String>::new();
 
@@ -792,6 +836,8 @@ impl NodeInfo {
             connections_key,
         } = params;
         let current_timestamp = Utc::now().timestamp_millis() as u64;
+        let operational_addresses: HashSet<String> =
+            operational_peer_wallets().await.into_iter().collect();
         if !is_public_network_address(&edit.ip) {
             return RpcResponse::Binary(b"Error: Invalid network address".to_vec());
         }
@@ -900,8 +946,14 @@ impl NodeInfo {
                         .unwrap_or(0);
                     let signer_blocks_mined =
                         signer_node.map(|node| node.blocks_mined).unwrap_or(0);
+                    let redirect = Self::sponsor_redirect_suffix(
+                        &address_map,
+                        &operational_addresses,
+                        current_timestamp,
+                        &signer_key,
+                    );
                     return RpcResponse::Binary(format!(
-                        "Error: This address cannot add nodes. It must exist for at least 60 minutes and not be marked for deletion local_height={local_height} self_add_block={} self_add_limit={} signer={signer_key} signer_exists={signer_exists} signer_is_local={signer_is_local} signer_deleted_timestamp={signer_deleted_timestamp} signer_age_ms={signer_age_ms} signer_blocks_mined={signer_blocks_mined}",
+                        "Error: This address cannot add nodes. It must exist for at least 60 minutes and not be marked for deletion local_height={local_height} self_add_block={} self_add_limit={} signer={signer_key} signer_exists={signer_exists} signer_is_local={signer_is_local} signer_deleted_timestamp={signer_deleted_timestamp} signer_age_ms={signer_age_ms} signer_blocks_mined={signer_blocks_mined}{redirect}",
                         super::SELF_ADD_BLOCK,
                         Self::self_add_limit_height(),
                     )
@@ -909,10 +961,15 @@ impl NodeInfo {
                 }
                 let mined_count = signer_node.map(|node| node.blocks_mined).unwrap_or(0);
                 if mined_count < 100 {
-                    return RpcResponse::Binary(
+                    let redirect = Self::sponsor_redirect_suffix(
-                        b"Error: This address cannot add nodes. It must mined 100 blocks before adding new nodes to the network"
+                        &address_map,
-                            .to_vec(),
+                        &operational_addresses,
+                        current_timestamp,
+                        &signer_key,
                     );
+                    return RpcResponse::Binary(format!(
+                        "Error: This address cannot add nodes. It must mine 100 blocks before adding new nodes to the network{redirect}"
+                    ).into_bytes());
                 }
             }
 
@@ -1122,6 +1179,69 @@ mod tests {
         }
     }
 
+    #[test]
+    fn sponsor_redirect_selects_an_eligible_different_node() {
+        let now = ONE_HOUR_MILLIS * 3;
+        let mut map = HashMap::new();
+
+        let mut rejected = snapshot_node("1.2.3.4", 1, &[]);
+        rejected.blocks_mined = 250;
+        map.insert("rejected.cltc".to_string(), rejected);
+
+        let mut newer = snapshot_node("1.2.3.6", ONE_HOUR_MILLIS, &[]);
+        newer.port = 50052;
+        newer.blocks_mined = 100;
+        map.insert("newer.cltc".to_string(), newer);
+
+        let mut older = snapshot_node("1.2.3.5", 2, &[]);
+        older.port = 50051;
+        older.blocks_mined = 100;
+        map.insert("older.cltc".to_string(), older);
+
+        assert_eq!(
+            NodeInfo::eligible_sponsor_endpoint_from_map(
+                &map,
+                &HashSet::from(["newer.cltc".to_string(), "older.cltc".to_string()]),
+                now,
+                "rejected.cltc",
+            ),
+            Some("1.2.3.5:50051".to_string())
+        );
+    }
+
+    #[test]
+    fn sponsor_redirect_ignores_deleted_young_and_under_mined_nodes() {
+        let now = ONE_HOUR_MILLIS * 3;
+        let mut map = HashMap::new();
+
+        let mut deleted = snapshot_node("1.2.3.4", 1, &[]);
+        deleted.blocks_mined = 100;
+        deleted.deleted_timestamp = now;
+        map.insert("deleted.cltc".to_string(), deleted);
+
+        let mut young = snapshot_node("1.2.3.5", now - 1_000, &[]);
+        young.blocks_mined = 100;
+        map.insert("young.cltc".to_string(), young);
+
+        let mut under_mined = snapshot_node("1.2.3.6", 1, &[]);
+        under_mined.blocks_mined = 99;
+        map.insert("under-mined.cltc".to_string(), under_mined);
+
+        assert_eq!(
+            NodeInfo::eligible_sponsor_endpoint_from_map(
+                &map,
+                &HashSet::from([
+                    "deleted.cltc".to_string(),
+                    "young.cltc".to_string(),
+                    "under-mined.cltc".to_string(),
+                ]),
+                now,
+                "none.cltc",
+            ),
+            None
+        );
+    }
+
     #[test]
     fn complete_snapshot_allows_membership_before_monitor_setup_finishes() {
         let mut map = HashMap::new();

src/records/memory/network_mapping/monitor.rs

@@ -106,6 +106,24 @@ mod tests {
         assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending));
     }
 
+    #[test]
+    fn only_freshly_sponsored_pending_membership_may_author_monitor_add() {
+        let address = "pending.cltc";
+        let mut pending = node(&[]);
+        pending.added_timestamp = 300;
+        pending.deleted_timestamp = 300;
+        pending.deleted_block = 0;
+
+        assert!(NodeInfo::can_pending_member_add_monitor(address, &pending));
+
+        pending.deleted_block = 50;
+        assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending));
+
+        pending.deleted_block = 0;
+        pending.added_by = address.to_string();
+        assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending));
+    }
+
     #[test]
     fn signed_monitor_wire_record_round_trips() {
         let monitored = "ab13318c26250b048db92920a80a86127c933b0c.cltc".to_string();
@@ -287,6 +305,13 @@ impl NodeInfo {
         edit.action == MONITOR_ACTION_ADD && edit.modified_timestamp > node.deleted_timestamp
     }
 
+    pub(crate) fn can_pending_member_add_monitor(address: &str, node: &NodeInfo) -> bool {
+        node.deleted_timestamp > 0
+            && node.deleted_timestamp == node.added_timestamp
+            && node.deleted_block == 0
+            && node.added_by != address
+    }
+
     pub(super) fn discard_old_target_ip_events_from(
         state: &mut HashMap<String, SignedMonitorEdit>,
         address: &str,
@@ -393,7 +418,10 @@ impl NodeInfo {
         let monitoring = address_map
             .get(&edit.monitoring_address)
             .ok_or_else(|| "monitoring address not found".to_string())?;
-        if monitoring.deleted_timestamp > 0 {
+        if monitoring.deleted_timestamp > 0
+            && !(edit.action == MONITOR_ACTION_ADD
+                && Self::can_pending_member_add_monitor(&edit.monitoring_address, monitoring))
+        {
             return Err("deleted node cannot update monitor relationships".to_string());
         }
 

src/records/memory/network_mapping/queries.rs

@@ -47,6 +47,18 @@ impl NodeInfo {
             .unwrap_or(false)
     }
 
+    pub(crate) async fn is_active_or_sponsored_pending(address: &str) -> bool {
+        ADDRESS_MAP
+            .lock()
+            .await
+            .get(address)
+            .map(|node| {
+                node.deleted_timestamp == 0
+                    || Self::can_pending_member_add_monitor(address, node)
+            })
+            .unwrap_or(false)
+    }
+
     pub(crate) async fn contains_address(address: &str) -> bool {
         ADDRESS_MAP.lock().await.contains_key(address)
     }
@@ -343,19 +355,6 @@ impl NodeInfo {
             .collect()
     }
 
-    pub async fn eligible_sponsor_ips() -> Vec<String> {
-        let current_timestamp = Utc::now().timestamp_millis() as u64;
-        let map = ADDRESS_MAP.lock().await;
-        map.values()
-            .filter(|node_info| {
-                node_info.deleted_timestamp == 0
-                    && current_timestamp.saturating_sub(node_info.added_timestamp) >= 3_600_000
-                    && node_info.blocks_mined >= 100
-            })
-            .map(|node_info| node_info.ip.clone())
-            .collect()
-    }
-
     pub async fn get_deleted_addresses() -> Vec<u8> {
         let map = ADDRESS_MAP.lock().await;
         map.iter()

src/rpc/client/handshake.rs

@@ -98,6 +98,28 @@ fn sponsor_endpoint_from_error(error: &str) -> Option<String> {
     Some(endpoint.to_string())
 }
 
+#[cfg(test)]
+mod tests {
+    use super::sponsor_endpoint_from_error;
+
+    #[test]
+    fn parses_server_sponsor_redirect() {
+        let error = "network self-announcement was rejected: Error: sponsor unavailable sponsor=1.2.3.4:50050";
+        assert_eq!(
+            sponsor_endpoint_from_error(error),
+            Some("1.2.3.4:50050".to_string())
+        );
+    }
+
+    #[test]
+    fn ignores_rejection_without_sponsor_redirect() {
+        assert_eq!(
+            sponsor_endpoint_from_error("network self-announcement was rejected"),
+            None
+        );
+    }
+}
+
 async fn connect_from_configured_ip(remote_addr: SocketAddr) -> io::Result<TcpStream> {
     let listen_ip = get_listen_ip().await;
     let local_ip: IpAddr = listen_ip.parse().map_err(|err| {

src/rpc/client/handshake_processing.rs

@@ -15,7 +15,7 @@ use crate::records::memory::connections::{
     spawn_peer_setup_retry, startup_synced_peer_streams, CONNECTIONS,
 };
 use crate::records::memory::network_mapping::NodeInfo;
-use crate::records::memory::response_channels::Command;
+use crate::records::memory::response_channels::{retire_peer_entries, Command};
 use crate::records::memory::structs::Connection;
 use crate::rpc::client::genesis_compat::ensure_compatible_genesis;
 use crate::rpc::client::governance_state_sync::sync_governance_state;
@@ -85,8 +85,15 @@ pub fn spawn_bootstrap_peer_discovery(params: BootstrapParams) {
 
     tokio::spawn(async move {
         let _startup_guard = startup_guard;
+        let failed_stream = params.stream.clone();
+        let failed_key = params.connections_key.clone();
+        let command_map = params.map.clone();
         if let Err(e) = bootstrap_peer_discovery(params).await {
-            eprintln!("[bootstrap] error: {e}");
+            warn!("[bootstrap] asynchronous peer setup failed for {failed_key}: {e}");
+            remove_stream_from_memory(&failed_stream).await;
+            // The stream may already have been removed by the failing setup
+            // path. Retire its request waiters explicitly in either case.
+            retire_peer_entries(command_map, &failed_key).await;
         }
     });
 }
@@ -558,13 +565,10 @@ pub async fn process_handshake_response(
         .map_err(|err| io::Error::other(format!("Network-map digest comparison failed: {err}")))?;
 
         if !matched {
-            info!(
+            info!("[network_map] reconciling signed mapping state with peer {connections_key}");
-                "[network_map] reconciling signed mapping state with peer {connections_key}"
-            );
             reconcile_network_mapping_with_peer(
                 broadcast_stream.clone(),
                 params.map.clone(),
-                &params.db,
                 &connections_key,
             )
             .await

src/rpc/server/handshake.rs

@@ -251,9 +251,27 @@ fn spawn_incoming_peer_promotion_watcher(
     });
 }
 
+fn needs_reverse_bootstrap_announcement(
+    self_add_allowed: bool,
+    local_membership_active: bool,
+) -> bool {
+    self_add_allowed && !local_membership_active
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{passive_catch_up_expired, peer_reached_catch_up_target};
+    use super::{
+        needs_reverse_bootstrap_announcement, passive_catch_up_expired,
+        peer_reached_catch_up_target,
+    };
+
+    #[test]
+    fn reverse_bootstrap_announcement_is_only_for_an_unregistered_first_node() {
+        assert!(needs_reverse_bootstrap_announcement(true, false));
+        assert!(!needs_reverse_bootstrap_announcement(true, true));
+        assert!(!needs_reverse_bootstrap_announcement(false, false));
+        assert!(!needs_reverse_bootstrap_announcement(false, true));
+    }
 
     #[test]
     fn moving_tip_does_not_move_the_original_catch_up_target() {
@@ -357,12 +375,16 @@ async fn complete_incoming_miner_setup(
         crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height(
             remote_height,
         );
-
-    if self_add_allowed {
-        // Before the sponsored-add gate, reverse announcement lets tiny early
-        // networks bootstrap from one live peer. After the gate activates, the
-        // incoming peer's outgoing join request is the only add path.
     let short_address = wallet.saved.short_address.clone();
+    let local_membership_active =
+        crate::records::memory::network_mapping::NodeInfo::is_active_address(&short_address).await;
+
+    if needs_reverse_bootstrap_announcement(self_add_allowed, local_membership_active) {
+        // Before the sponsored-add gate, reverse announcement lets tiny early
+        // networks bootstrap from one live peer. It is only valid while this
+        // receiving node has no active membership of its own. An established
+        // node must never import even its own record from a newly connected
+        // peer.
         if let Err(err) = announce_self_to_network(
             stream.clone(),
             &short_address,

src/startup/connections.rs

@@ -1,3 +1,4 @@
+use crate::common::check_genesis::genesis_checkup;
 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};
@@ -39,6 +40,11 @@ pub async fn handle_connections(
     wallet: Arc<Wallet>,
     map: Arc<Mutex<Command>>,
 ) -> Result<(), String> {
+    // Reciprocal sponsorship is only the bootstrap rule for the first two
+    // nodes that are establishing a chain. A node restarting an existing
+    // chain may connect to any valid operational peer.
+    let first_two_node_startup = !genesis_checkup().await;
+
     // A zero outgoing limit means this node waits for an incoming sponsor. It
     // does not bypass the independent-peer requirement or permit solo mining.
     let outgoing_connections = crate::Settings::load()
@@ -56,17 +62,22 @@ pub async fn handle_connections(
 
     // Startup gets one bounded pass through configured peers. If none can
     // sponsor this node, wait for a real incoming peer instead of repeatedly
-    // dialing the list. Genesis remains blocked until reciprocal sponsorship
+    // dialing the list. Only the first two nodes establishing genesis require
-    // makes that peer operational.
+    // reciprocal sponsorship; mature restarts need an operational peer.
     info!("No existing network peer is available. Waiting for connections.");
     loop {
         let peer_wallets = operational_peer_wallets().await;
-        if is_normal_mode()
+        let sponsorship_ready = if first_two_node_startup {
-            && !is_mining_stop_requested()
+            NodeInfo::has_reciprocal_sponsorship(&wallet.saved.short_address, &peer_wallets).await
-            && NodeInfo::has_reciprocal_sponsorship(&wallet.saved.short_address, &peer_wallets)
+        } else {
-                .await
+            !peer_wallets.is_empty()
-        {
+        };
+        if is_normal_mode() && !is_mining_stop_requested() && sponsorship_ready {
+            if first_two_node_startup {
                 info!("Reciprocal node sponsorship completed; startup may continue.");
+            } else {
+                info!("Operational peer connected; existing-chain startup may continue.");
+            }
             return Ok(());
         }
         sleep(Duration::from_secs(1)).await;

src/startup/network_broadcast.rs

@@ -140,15 +140,6 @@ async fn announce_self_to_network_inner(
     Ok(())
 }
 
-pub async fn get_network_mapping(
-    unlocked_stream: Arc<Mutex<TcpStream>>,
-    command_map: Arc<Mutex<Command>>,
-    db: &Db,
-    connections_key: &str,
-) -> Result<(), String> {
-    get_network_mapping_inner(unlocked_stream, command_map, db, connections_key, None).await
-}
-
 pub async fn compare_network_mapping_digest(
     unlocked_stream: Arc<Mutex<TcpStream>>,
     command_map: Arc<Mutex<Command>>,
@@ -196,7 +187,9 @@ async fn send_mapping_record(
         .ok_or_else(|| "network-map reconciliation response channel closed".to_string())?;
     let response = binary_to_string(response);
     if response != "Success" {
-        return Err(format!("peer rejected reconciled mapping record: {response}"));
+        return Err(format!(
+            "peer rejected reconciled mapping record: {response}"
+        ));
     }
     Ok(())
 }
@@ -204,7 +197,6 @@ async fn send_mapping_record(
 pub async fn reconcile_network_mapping_with_peer(
     unlocked_stream: Arc<Mutex<TcpStream>>,
     command_map: Arc<Mutex<Command>>,
-    db: &Db,
     connections_key: &str,
 ) -> Result<(), String> {
     let (memberships, monitor_events) = NodeInfo::signed_mapping_state().await;
@@ -218,7 +210,9 @@ pub async fn reconcile_network_mapping_with_peer(
         let signature = crate::decode(&edit.modified_signature)
             .map_err(|_| "local mapping contained an invalid membership signature".to_string())?;
         if signature.len() != Wallet::SIGNATURE_LENGTH {
-            return Err("local mapping contained an invalid membership signature length".to_string());
+            return Err(
+                "local mapping contained an invalid membership signature length".to_string(),
+            );
         }
 
         let mut body = Vec::new();
@@ -279,10 +273,12 @@ pub async fn reconcile_network_mapping_with_peer(
         .await?;
     }
 
-    // The peer has now merged every valid current signed record we hold. Pull
+    // Reconciliation on an additional/discovered connection is intentionally
-    // its resulting complete state so records that only it possessed are also
+    // one-way. This node already joined through its bootstrap owner, so the
-    // installed locally and derived deletion cascades are identical.
+    // newly connected peer may merge our signed state but must never replace
-    get_network_mapping(unlocked_stream, command_map, db, connections_key).await
+    // this node's established mapping. Missing records continue to arrive
+    // through the normal redundant network broadcasts.
+    Ok(())
 }
 
 pub async fn get_network_mapping_for_address(