summaryrefslogtreecommitdiffstats
path: root/src/bootstrap.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/bootstrap.rs')
-rw-r--r--src/bootstrap.rs104
1 files changed, 0 insertions, 104 deletions
diff --git a/src/bootstrap.rs b/src/bootstrap.rs
deleted file mode 100644
index d3ac94d..0000000
--- a/src/bootstrap.rs
+++ /dev/null
@@ -1,104 +0,0 @@
-//! 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;
- }
-}