Function handle_join_request

nodedb-cluster/src/bootstrap/handle_join.rs:39–115

Build a `JoinResponse` for an incoming `JoinRequest`. See module docs for semantics. Mutates `topology` only when the node is newly admitted; idempotent for re-joins with the same address. `cluster_id` is the id of the cluster this node belongs to — the join flow reads it from the local catalog and threads it through so the joining node can persist it and take the `restart()` path on a subsequen

(
    req: &JoinRequest,
    topology: &mut ClusterTopology,
    routing: &RoutingTable,
    cluster_id: u64,
)

/// catalog has not yet been populated; rejection responses also carry
/// zero.
pub fn handle_join_request(
    req: &JoinRequest,
    topology: &mut ClusterTopology,
    routing: &RoutingTable,
    cluster_id: u64,
) -> JoinResponse {
    // Validate the wire version carried in the JOIN payload (belt-and-suspenders
    // check; the transport-level handshake already negotiated a compatible version
    // before this RPC was dispatched). The `wire_version` field here is the
    // cluster-wide schema version (`CLUSTER_WIRE_FORMAT_VERSION`), distinct from
    // the transport-level RPC frame version. We require an exact match because
    // this build uses floor == ceiling (no backward-compat window in the schema).
    if req.wire_version != CLUSTER_WIRE_FORMAT_VERSION {
        warn!(
            node_id = req.node_id,
            joiner_wire_version = req.wire_version,
            expected_wire_version = CLUSTER_WIRE_FORMAT_VERSION,
            "join request rejected: joiner cluster wire_version mismatch"
        );
        return reject(format!(
            "joiner wire_version {} does not match this cluster's wire_version {} — \
             rolling upgrade is required before this node can join",
            req.wire_version, CLUSTER_WIRE_FORMAT_VERSION
        ));
    }

    // Validate the listen address early.
    let addr: SocketAddr = match req.listen_addr.parse() {
        Ok(a) => a,
        Err(e) => {
            return reject(format!("invalid listen_addr '{}': {e}", req.listen_addr));
        }
    };

    // Collision / idempotency check — both require reading the existing entry.
    if let Some(existing) = topology.get_node(req.node_id) {
        let existing_addr = existing.addr.clone();
        if existing_addr != req.listen_addr {
            // Same id, different address — reject.
            return reject(format!(
                "node_id {} already registered with different address {} (request: {})",
                req.node_id, existing_addr, req.listen_addr
            ));
        }
        // Same id, same address. If already Active we short-circuit —
        // no topology mutation at all, just rebuild the wire response. If the
        // node was in a non-Active state (Joining/Draining), normalize to
        // Active now because it's clearly back online.
        if existing.state != NodeState::Active
            && let Some(entry) = topology.get_node_mut(req.node_id)
        {
            entry.state = NodeState::Active;
        }
        return build_response(topology, routing, cluster_id);
    }

    // Brand new node — admit as Active. Stamp the joiner's own
    // wire version and identity fields onto its NodeInfo so every