Method spawn_inner

nodedb-cluster/tests/common/test_node.rs:161–287

(
        node_id: u64,
        transport: Arc<NexarTransport>,
        seed_nodes: Vec<SocketAddr>,
        data_dir_path: PathBuf,
        owned_data_dir: Option<TempDir>,
    )

    }

    pub async fn spawn_inner(
        node_id: u64,
        transport: Arc<NexarTransport>,
        seed_nodes: Vec<SocketAddr>,
        data_dir_path: PathBuf,
        owned_data_dir: Option<TempDir>,
    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
        let catalog = Arc::new(ClusterCatalog::open(&data_dir_path.join("cluster.redb"))?);
        let listen_addr = transport.local_addr();

        // Empty seeds → imply single-node bootstrap by listing only
        // our own address. Otherwise use whatever the caller supplied.
        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,
            // Fast retry policy: 2 s ceiling keeps the join-failure
            // tests (especially `cluster_join_leader_crash`) under
            // ~5 s of sleeping instead of the production ~64 s.
            join_retry: nodedb_cluster::JoinRetryPolicy {
                max_attempts: 8,
                max_backoff_secs: 2,
            },
            swim_udp_addr: None,
            election_timeout_min: std::time::Duration::from_millis(150),
            election_timeout_max: std::time::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?;
        // Match the main binary: the caller is responsible for the
        // final `Ready` transition once the node is wired up. The
        // node count is whatever the node observed at the moment of
        // transition.
        lifecycle.to_ready(state.topology.read().map(|t| t.node_count()).unwrap_or(0));

        // state.topology is already Arc<RwLock<ClusterTopology>>.
        let topology = state.topology.clone();
        // Real in-memory metadata cache, driven by a `CacheApplier`
        // installed on the raft loop. Every test can read this
        // directly to assert DDL replication.
        let metadata_cache = Arc::new(RwLock::new(MetadataCache::new()));
        let metadata_applier: Arc<dyn nodedb_cluster::MetadataApplier> =
            Arc::new(CacheApplier::new(metadata_cache.clone()));
        // `start_cluster` no longer spawns subsystems, so its
        // `Arc<Mutex<MultiRaft>>` has exactly one strong owner here.