Function join

nodedb-cluster/src/bootstrap/join.rs:74–139

Join an existing cluster by contacting seed nodes. The loop has two layers: - **Outer**: retry passes with exponential backoff per `config.join_retry`. Handles the "bootstrapper not up yet" startup race. - **Inner**: walk the seed list plus any leader-redirect hops for this attempt. A successful `JoinResponse` short-circuits the whole function; failures on one candidate fall through to the next.

(
    config: &ClusterConfig,
    catalog: &ClusterCatalog,
    transport: &NexarTransport,
    lifecycle: &ClusterLifecycleTracker,
)

///   whole function; failures on one candidate fall through to the
///   next.
pub(super) async fn join(
    config: &ClusterConfig,
    catalog: &ClusterCatalog,
    transport: &NexarTransport,
    lifecycle: &ClusterLifecycleTracker,
) -> Result<ClusterState> {
    info!(
        node_id = config.node_id,
        seeds = ?config.seed_nodes,
        "joining existing cluster"
    );

    if config.seed_nodes.is_empty() {
        let err = ClusterError::Transport {
            detail: "no seed nodes configured".into(),
        };
        lifecycle.to_failed(err.to_string());
        return Err(err);
    }

    let req_template = JoinRequest {
        node_id: config.node_id,
        listen_addr: config.listen_addr.to_string(),
        wire_version: crate::topology::CLUSTER_WIRE_FORMAT_VERSION,
        spiffe_id: None,
        spki_pin: transport.local_spki_pin().map(|arr| arr.to_vec()),
    };

    let policy = config.join_retry;
    let mut last_err: Option<ClusterError> = None;

    for attempt in 0..policy.max_attempts {
        lifecycle.to_joining(attempt);

        let delay = policy.backoff_for(attempt);
        if !delay.is_zero() {
            debug!(
                node_id = config.node_id,
                attempt,
                delay_ms = delay.as_millis() as u64,
                "backing off before next join attempt"
            );
            tokio::time::sleep(delay).await;
        }

        match try_join_once(config, catalog, transport, &req_template).await {
            Ok(state) => return Ok(state),
            Err(e) => {
                warn!(
                    node_id = config.node_id,
                    attempt,
                    error = %e,
                    "join attempt failed; will retry"
                );
                last_err = Some(e);
            }
        }
    }