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
|
//! Joining a cluster (or creating one as the genesis node).
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::{bail, Result};
use tracing::{debug, error, info, warn};
use uuid::Uuid;
use crate::peer::{remember_peer, send_hello_once};
use crate::state::{AppState, PeerMembership};
const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300);
const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500);
pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> {
if seeds.is_empty() {
info!("Initializing genesis node (no seed peers configured)");
state.initial_sync_done.store(true, Ordering::Release);
state.mark_ready(Uuid::new_v4().to_string()).await?;
return Ok(());
}
info!(
timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
"Joining existing cluster"
);
let mut peers = Vec::new();
for seed in seeds {
match state.transport.resolve_peer(seed).await {
Ok(peer) => {
debug!(seed = %seed, peer = %peer, "Resolved seed to peer");
peers.push(peer)
}
Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"),
}
}
if peers.is_empty() {
bail!("None of the configured seeds could be resolved");
}
for peer in peers {
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");
}
}
let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT;
loop {
if let Some(cluster_id) = state.cluster_id().await {
// A node joining a compacted cluster rejoins the newer epoch first; it is
// ready only once that rejoin verified the epoch.
if state.initial_sync_done.load(Ordering::Acquire) && !state.compaction.is_rejoining() {
state.mark_ready(cluster_id.clone()).await?;
info!(cluster_id = %cluster_id, "Node is ready");
return Ok(());
}
}
if tokio::time::Instant::now() >= deadline {
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;
let validated_cluster = match &peer_state.membership {
PeerMembership::Unknown => None,
PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()),
};
peer_diagnostics.push(format!(
"{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}",
validated_cluster,
peer_state.epoch,
peer_state.pending_inbound_sync.len()
))
}
let local_cluster = state.cluster_id().await;
let local_epoch = state.local_epoch();
let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire);
error!(
local_cluster = ?local_cluster,
local_epoch = %local_epoch,
initial_sync_done,
peers = %peer_diagnostics.join("; "),
timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(),
"Bootstrap timed out"
);
bail!(
"Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.",
BOOTSTRAP_TIMEOUT,
local_cluster,
local_epoch,
initial_sync_done,
peer_diagnostics.join("; ")
);
}
tokio::time::sleep(BOOTSTRAP_RETRY).await;
}
}
|
