Function apply_join_response

nodedb-cluster/src/bootstrap/join.rs:242–359

Apply a JoinResponse: reconstruct topology, routing, and MultiRaft from wire data. Order of operations is load-bearing for crash safety: 1. Reconstruct the `ClusterTopology` and `RoutingTable` in memory. 2. Persist topology + routing to the catalog **first**, before any on-disk side effects. If we crash after this step, the next boot sees `catalog.is_bootstrapped() == true` and takes the `restar

(
    config: &ClusterConfig,
    catalog: &ClusterCatalog,
    transport: &NexarTransport,
    resp: &JoinResponse,
)

/// 4. Register every peer address in the transport before returning
///    so the first outgoing AppendEntries has a known destination.
fn apply_join_response(
    config: &ClusterConfig,
    catalog: &ClusterCatalog,
    transport: &NexarTransport,
    resp: &JoinResponse,
) -> Result<ClusterState> {
    // 1. Reconstruct topology.
    let mut topology = ClusterTopology::new();
    for node in &resp.nodes {
        let state = NodeState::from_u8(node.state).unwrap_or(NodeState::Active);
        let spki_pin: Option<[u8; 32]> = node.spki_pin.as_deref().and_then(|b| {
            if b.len() == 32 {
                let mut arr = [0u8; 32];
                arr.copy_from_slice(b);
                Some(arr)
            } else {
                None
            }
        });
        let mut info = NodeInfo::new(
            node.node_id,
            node.addr.parse().unwrap_or_else(|_| {
                "0.0.0.0:0"
                    .parse()
                    .expect("invariant: \"0.0.0.0:0\" is a valid SocketAddr literal")
            }),
            state,
        )
        .with_wire_version(node.wire_version)
        .with_spiffe_id(node.spiffe_id.clone())
        .with_spki_pin(spki_pin);
        // Override raft_groups from wire data (NodeInfo::new starts empty).
        info.raft_groups = node.raft_groups.clone();
        if node.node_id == config.node_id {
            info.state = NodeState::Active;
        }
        topology.add_node(info);
    }

    // 1. Reconstruct routing table.
    let mut group_members = std::collections::HashMap::new();
    for g in &resp.groups {
        group_members.insert(
            g.group_id,
            GroupInfo {
                leader: g.leader,
                members: g.members.clone(),
                learners: g.learners.clone(),
            },
        );
    }
    let routing = RoutingTable::from_parts(resp.vshard_to_group.clone(), group_members);

    // 2. Persist to catalog before any on-disk Raft side effects.
    //    Cluster id is written first so `is_bootstrapped()` returns
    //    `true` on any subsequent boot — without this, a joined node
    //    that restarts would re-enter the bootstrap/join path
    //    instead of taking `restart()`. Zero is a valid marker: the