diff options
Diffstat (limited to 'src/bootstrap.rs')
| -rw-r--r-- | src/bootstrap.rs | 104 |
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; - } -} |
