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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
// src/network/bootstrap.rs
use super::{ConnectionPolicy, 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 connection_policy = ConnectionPolicy::from_network_config(config);
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,
);
break;
}
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);
}
if let Ok(sock) = addr_clone.parse() {
peer_store_clone.mark_failure(&sock).await;
let Some(attempt) = peer_store_clone
.next_reconnect_attempt(
&sock,
connection_policy.reconnect_max_attempts,
)
.await
else {
emit_network_event(
"bootstrap",
LogLevel::Warn,
"bootstrap_reconnect_exhausted",
Some(addr_clone.clone()),
None,
allow_console,
);
break;
};
tokio::time::sleep(
connection_policy.reconnect_delay_with_random_jitter(attempt),
)
.await;
} else {
break;
}
}
}
// On any outcome, persist latest store periodically if enabled
peer_store_clone.save_if_enabled(&config_clone).await;
}
});
}
} else {
emit_network_event(
"bootstrap",
LogLevel::Info,
"bootstrap_nodes_missing",
None,
Some("source=config".to_string()),
allow_console,
);
}
}