thenodes 0.2.0

TheNodes is a modular, plugin-driven P2P node framework for Rust, supporting node-embedded plugins (NEP) and core-as-a-library (CAL) modes with async-first APIs.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

// src/network/bootstrap.rs

use super::{Peer, PeerSource, PeerStore};
use crate::config::Config;
use crate::events::model::LogLevel;
use crate::network::events::emit_network_event;
use crate::network::peer_manager::PeerManager;
use crate::network::transport::connect_to_peer;
use crate::plugin_host::manager::PluginManager;
use crate::realms::RealmInfo;
use std::sync::Arc;
use tokio::sync::Mutex as TokioMutex;

#[allow(clippy::too_many_arguments)]
pub async fn connect_to_bootstrap_nodes(
    config: &Config,
    realm: RealmInfo,
    peer_manager: PeerManager,
    plugin_manager: Arc<PluginManager>,
    error_buffer: Arc<TokioMutex<Vec<String>>>,
    allow_console: bool,
    local_node_id: String,
    peer_store: PeerStore,
) {
    if let Some(peers) = &config.bootstrap_nodes {
        let node_id_arc = Arc::new(local_node_id);
        for addr in peers {
            let peer = Peer::new("bootstrap".to_string(), addr.clone());
            let realm_clone = realm.clone();
            let addr_clone = addr.clone();
            let peer_manager_clone = peer_manager.clone();
            let port = config.port;
            let plugin_manager_clone = plugin_manager.clone();
            let error_buffer = error_buffer.clone();
            let config_clone = config.clone();
            let node_id_for_task = node_id_arc.clone();
            let peer_store_clone = peer_store.clone();
            // Seed into store
            if let Ok(sock) = addr.parse() {
                peer_store_clone.insert(sock, PeerSource::Bootstrap).await;
            }
            tokio::spawn(async move {
                loop {
                    // Suppress outbound dial if we already have a peer advertising this listen address
                    if peer_manager_clone.has_listen_addr(&addr_clone).await {
                        // Longer backoff when suppressed
                        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
                        continue;
                    }
                    match connect_to_peer(crate::network::transport::ConnectToPeerParams {
                        peer: &peer,
                        our_realm: realm_clone.clone(),
                        our_port: port,
                        peer_manager: peer_manager_clone.clone(),
                        plugin_manager: plugin_manager_clone.clone(),
                        allow_console,
                        config: config_clone.clone(),
                        local_node_id: (*node_id_for_task).clone(),
                        peer_store: Some(peer_store_clone.clone()),
                    })
                    .await
                    {
                        Ok(_) => {
                            emit_network_event(
                                "bootstrap",
                                LogLevel::Info,
                                "bootstrap_connect_success",
                                Some(addr_clone.clone()),
                                None,
                                allow_console,
                            );
                        }
                        Err(e) => {
                            emit_network_event(
                                "bootstrap",
                                LogLevel::Warn,
                                "bootstrap_connect_failed",
                                Some(addr_clone.clone()),
                                Some(e.to_string()),
                                allow_console,
                            );
                            let msg = format!("❌ Failed to connect to {}: {}", addr_clone, e);
                            #[allow(unused_must_use)]
                            {
                                error_buffer.lock().await.push(msg);
                            }
                        }
                    }
                    // On any outcome, persist latest store periodically if enabled
                    peer_store_clone.save_if_enabled(&config_clone).await;
                    // Wait before retrying
                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                    // Optionally buffer retry message as well
                }
            });
        }
    } else {
        emit_network_event(
            "bootstrap",
            LogLevel::Info,
            "bootstrap_nodes_missing",
            None,
            Some("source=config".to_string()),
            allow_console,
        );
    }
}