Function spawn_one_calvin_node

nodedb-cluster/tests/common/calvin_test_node.rs:204–328

Spawn a single `CalvinTestNode`. `all_node_ids` is the full 3-node voter set; `node_idx` is the index of this node in that list. Registers the sequencer group **before** spawning background Raft tasks. `seed_nodes` must be empty for the bootstrap node (index 0) and contain the bootstrap node's address for nodes 1 and 2.

(
    node_idx: usize,
    all_node_ids: &[u64],
    seed_nodes: Vec<SocketAddr>,
)

/// `seed_nodes` must be empty for the bootstrap node (index 0) and contain
/// the bootstrap node's address for nodes 1 and 2.
async fn spawn_one_calvin_node(
    node_idx: usize,
    all_node_ids: &[u64],
    seed_nodes: Vec<SocketAddr>,
) -> Result<CalvinTestNode, Box<dyn std::error::Error + Send + Sync>> {
    let node_id = all_node_ids[node_idx];
    let transport = Arc::new(NexarTransport::with_timeout(
        node_id,
        "127.0.0.1:0".parse().unwrap(),
        Duration::from_secs(4),
        TransportCredentials::Insecure,
    )?);

    let data_dir = tempfile::tempdir()?;
    let data_dir_path = data_dir.path().to_path_buf();
    let catalog = Arc::new(ClusterCatalog::open(&data_dir_path.join("cluster.redb"))?);

    let listen_addr = transport.local_addr();
    let seeds = if seed_nodes.is_empty() {
        vec![listen_addr]
    } else {
        seed_nodes
    };

    let config = ClusterConfig {
        node_id,
        listen_addr,
        seed_nodes: seeds,
        num_groups: 2,
        replication_factor: 3,
        data_dir: data_dir_path.clone(),
        force_bootstrap: false,
        join_retry: nodedb_cluster::JoinRetryPolicy {
            max_attempts: 8,
            max_backoff_secs: 2,
        },
        swim_udp_addr: None,
        election_timeout_min: Duration::from_millis(150),
        election_timeout_max: Duration::from_millis(300),
        install_snapshot_chunk_bytes: 4 * 1024 * 1024,
        orphan_partial_max_age_secs: 300,
    };

    let lifecycle = ClusterLifecycleTracker::new();
    let state = start_cluster(&config, &catalog, Arc::clone(&transport), &lifecycle).await?;
    lifecycle.to_ready(state.topology.read().map(|t| t.node_count()).unwrap_or(0));

    // Unwrap MultiRaft immediately — no other Arc clones of it yet.
    let mut multi_raft_value = Arc::try_unwrap(state.multi_raft)
        .unwrap_or_else(|_| {
            panic!("MultiRaft should have no extra strong Arc owners in Calvin test setup")
        })
        .into_inner()
        .unwrap_or_else(|p| p.into_inner());

    // Register the sequencer group before spawning any tasks.
    // This is the race-fix: we do this synchronously here so the 150 ms
    // election timeout cannot fire before the group is registered on