Contractless/src/startup/connections.rs

use crate::common::network_startup::get_node_connections;
use crate::log::{error, info, warn};
use crate::miner::flag::{
    clear_mining_stop_request, is_normal_mode, set_mining_state, set_node_mode, MiningState,
    NodeMode,
};
use crate::records::memory::connections::{
    peer_connection_count, ready_outgoing_connection_count, refill_outgoing_connections_once,
};
use crate::records::memory::response_channels::Command;
use crate::rpc::client::handshake::connect_and_handshake;
use crate::rpc::client::structs::Connect;
use crate::sled::Db;
use crate::sleep;
use crate::wallets::structures::Wallet;
use crate::Arc;
use crate::Duration;
use crate::Mutex;

pub async fn handle_connections(
    db: Db,
    wallet: Arc<Wallet>,
    map: Arc<Mutex<Command>>,
) -> Result<(), String> {
    // A zero outgoing limit means this node should not open any bootstrap
    // connection during startup.
    let outgoing_connections = crate::Settings::load()
        .map(|settings| settings.outgoing_connections)
        .unwrap_or(0);
    if outgoing_connections == 0 {
        info!("OUTGOING_CONNECTIONS is 0; skipping startup bootstrap.");
        set_node_mode(NodeMode::Normal);
        clear_mining_stop_request();
        set_mining_state(MiningState::Idle);
        return Ok(());
    }

    let connected = attempt_bootstrap_connections(db, wallet, map, "startup").await?;
    if connected {
        return Ok(());
    }

    // Startup can continue as a standalone node even if no bootstrap peer is reachable.
    set_node_mode(NodeMode::Normal);
    clear_mining_stop_request();
    set_mining_state(MiningState::Idle);
    Ok(())
}

async fn attempt_bootstrap_connections(
    db: Db,
    wallet: Arc<Wallet>,
    map: Arc<Mutex<Command>>,
    context: &str,
) -> Result<bool, String> {
    // Try the configured bootstrap peers one by one until a
    // handshake succeeds or the list is exhausted.
    let filtered_servers = get_node_connections().await;
    let mut last_error: Option<String> = None;

    for server in filtered_servers {
        // build the outbound handshake request using cloned
        // shared state so each attempt can run independently
        let db_clone = db.clone();

        // parse the configured peer string once before spawning
        // the outbound connection attempt
        let socket_address = server.parse().expect("Failed to parse the socket address");

        // Clone the Arc for use in other async functions
        let map_clone = Arc::clone(&map);

        let first: bool = true;
        let connect_params = Connect {
            addr: socket_address,
            db: db_clone,
            node_ip: server.to_string(),
            wallet: wallet.clone(),
            map: map_clone,
            first,
        };

        let err_string = match connect_and_handshake(connect_params).await {
            Ok(()) => {
                if context == "startup" {
                    info!("Connected to {server}");
                } else {
                    info!("[reconnect] bootstrap recovery connected to {server}");
                }
                return Ok(true);
            }
            Err(err) => err.to_string(),
        };

        // During outage recovery, a peer may already have the connection
        // recorded from an overlapping reconnect path. That is not a fatal
        // bootstrap-recovery condition; try the next configured peer.
        if err_string.contains(
            "The connection is already in the connection manager Please wait 10 minutes and try again",
        ) {
            if context == "startup" {
                error!("Error connecting to {server}: {err_string}");
            } else {
                warn!(
                    "[reconnect] bootstrap recovery skipped duplicate peer {server}: {err_string}"
                );
            }
            last_error = Some(err_string.clone());
            continue;
        }
        if context == "startup" {
            error!("Error connecting to {server}: {err_string}");
        } else {
            warn!("[reconnect] bootstrap recovery failed to connect to {server}: {err_string}");
        }
        last_error = Some(err_string.clone());
        sleep(Duration::from_secs(5)).await;
    }

    if let Some(err) = last_error {
        if context == "startup" {
            info!("No bootstrap peers connected during startup: {err}");
        } else {
            warn!("[reconnect] bootstrap recovery found no peers: {err}");
        }
    } else {
        if context == "startup" {
            info!("No bootstrap peers connected during startup.");
        } else {
            warn!("[reconnect] bootstrap recovery has no configured peers to try");
        }
    }

    Ok(false)
}

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)
        .unwrap_or(0);
    if outgoing_connections == 0 {
        return;
    }

    tokio::spawn(async move {
        loop {
            sleep(Duration::from_secs(60)).await;

            if !is_normal_mode() {
                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}"),
                }
                continue;
            }

            // A topology pass is bounded: it walks the current active map once,
            // tries each eligible endpoint at most once, then sleeps until the
            // next interval even when the configured limit exceeds network size.
            if ready_outgoing_connection_count().await < outgoing_connections as usize {
                refill_outgoing_connections_once().await;
            }
        }
    });
}