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
133
134
135
136
137
|
//! 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.
//!
//! The genesis node records the swarm ID in the document (`_continuum/swarm`), so
//! snapshots carry it forward and every backup names its swarm.
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::{Context, Result};
use serde_json::Value as JsonValue;
use tokio::time::Instant;
use tracing::{debug, info, warn};
use uuid::Uuid;
use crate::document::store;
use crate::remake::snapshot::{META_ROOT, SWARM};
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 {
let swarm_id = Uuid::new_v4().to_string();
info!(swarm_id = %swarm_id, "Starting a new swarm as its genesis node (--genesis)");
let recorded = JsonValue::String(swarm_id.clone());
state
.mutate_doc(move |doc| store::write_path(doc, &swarm_id_path(), &recorded))
.await
.context("Recording the swarm ID")?;
state.initial_sync_done.store(true, Ordering::Release);
state.mark_ready(swarm_id).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;
}
}
fn swarm_id_path() -> Vec<String> {
vec![META_ROOT.to_owned(), SWARM.to_owned()]
}
/// 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"
);
}
|
