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
|
//! Joining a swarm (or creating one as the genesis node).
//!
//! A node starts a new swarm only with `--genesis`. Otherwise it joins through its
//! seeds and keeps trying until it is ready: seeds that don't resolve are resolved
//! again with backoff, and HELLOs to resolved ones are retried by the peer worker. A
//! node without seeds waits (nothing to join yet) and says so now and then.
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Result;
use tokio::time::Instant;
use tracing::{debug, info, warn};
use uuid::Uuid;
use crate::state::AppState;
use crate::swarm::peer::{remember_peer, send_hello_once};
const BOOTSTRAP_POLL: Duration = Duration::from_millis(500);
/// How often a node that is still joining says why.
const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300);
/// Retrying seeds that didn't resolve: first after this, doubling up to the maximum.
const RESOLVE_RETRY: Duration = Duration::from_secs(5);
const RESOLVE_RETRY_MAX: Duration = Duration::from_secs(300);
pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) -> Result<()> {
if genesis {
info!("Starting a new swarm as its genesis node (--genesis)");
state.initial_sync_done.store(true, Ordering::Release);
state.mark_ready(Uuid::new_v4().to_string()).await?;
return Ok(());
}
if seeds.is_empty() {
warn!(
"No seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one"
);
} else {
info!(seeds = seeds.len(), "Joining existing swarm");
}
let mut unresolved: Vec<String> = seeds.to_vec();
let mut resolve_retry = RESOLVE_RETRY;
let mut next_resolve = Instant::now();
let mut next_report = Instant::now() + BOOTSTRAP_REPORT;
loop {
if let Some(swarm_id) = state.swarm_id().await {
// A node joining a compacted swarm rejoins the newer epoch first; it is
// ready only once that rejoin verified the epoch.
if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
state.mark_ready(swarm_id.clone()).await?;
info!(swarm_id = %swarm_id, "Node is ready");
return Ok(());
}
}
let now = Instant::now();
if !unresolved.is_empty() && now >= next_resolve {
resolve_seeds(state, &mut unresolved).await;
if !unresolved.is_empty() {
next_resolve = now + resolve_retry;
resolve_retry = (resolve_retry * 2).min(RESOLVE_RETRY_MAX);
}
}
if now >= next_report {
next_report = now + BOOTSTRAP_REPORT;
report_joining(state, seeds, &unresolved).await;
}
tokio::time::sleep(BOOTSTRAP_POLL).await;
}
}
/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
/// rest for later.
async fn resolve_seeds(state: &AppState, unresolved: &mut Vec<String>) {
let mut still = Vec::new();
for seed in unresolved.drain(..) {
match state.transport.resolve_seed(&seed).await {
Ok(peer) => {
debug!(seed = %seed, peer = %peer, "Resolved seed");
remember_peer(state, &peer).await;
if let Err(error) = send_hello_once(state, &peer).await {
warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry");
}
}
Err(error) => {
warn!(seed = %seed, error = %error, "Seed doesn't resolve; will retry");
still.push(seed);
}
}
}
*unresolved = still;
}
/// Why the node is still joining, for the log.
async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String]) {
if seeds.is_empty() {
warn!(
"Still no seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one"
);
return;
}
let mut peer_diagnostics: Vec<String> = Vec::new();
let peer_states = state.peer_states.read().await.clone();
for (peer, peer_state_mutex) in &peer_states {
let peer_state = peer_state_mutex.lock().await;
peer_diagnostics.push(format!(
"{peer}: validated_swarm={:?}, epoch={:?}",
peer_state.standing.swarm_id(),
peer_state.epoch,
));
}
warn!(
local_swarm = ?state.swarm_id().await,
local_epoch = %state.local_epoch(),
initial_sync_done = state.initial_sync_done.load(Ordering::Acquire),
unresolved_seeds = ?unresolved,
peers = %peer_diagnostics.join("; "),
"Still joining the swarm; trying on"
);
}
|
