1 files changed, 104 insertions, 0 deletions
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
new file mode 100644
index 0000000..2907e93
--- /dev/null
+++ b/src/swarm/bootstrap.rs
@@ -0,0 +1,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::state::{AppState, PeerMembership};
+use crate::swarm::peer::{remember_peer, send_hello_once};
+
+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;
+    }
+}