Function start_cluster_subsystems

nodedb-cluster/src/bootstrap/start.rs:183–215

Spawn the default cluster subsystems sharing `raft_multi_raft` with the running [`crate::raft_loop::RaftLoop`]. Called after [`start_cluster`] has produced [`ClusterState`] and the host has handed `MultiRaft` over to the `RaftLoop`. The host passes `raft_multi_raft = raft_loop.multi_raft_handle()` here so subsystems use the same `Arc >` the loop owns — no double-ownership, no orph

(
    config: &ClusterConfig,
    topology: Arc<std::sync::RwLock<crate::topology::ClusterTopology>>,
    routing: Arc<std::sync::RwLock<crate::routing::RoutingTable>>,
    transport: Arc<NexarTranspo

/// orderly shutdown so subsystems release their `MultiRaft` Arc
/// before the loop exits.
pub async fn start_cluster_subsystems(
    config: &ClusterConfig,
    topology: Arc<std::sync::RwLock<crate::topology::ClusterTopology>>,
    routing: Arc<std::sync::RwLock<crate::routing::RoutingTable>>,
    transport: Arc<NexarTransport>,
    raft_multi_raft: Arc<std::sync::Mutex<crate::multi_raft::MultiRaft>>,
) -> Result<RunningCluster> {
    let health = ClusterHealth::new();
    let ctx = BootstrapCtx::new(
        Arc::clone(&topology),
        Arc::clone(&routing),
        Arc::clone(&transport),
        Arc::clone(&raft_multi_raft),
        health,
    );

    let executor = Arc::new(MigrationExecutor::new(
        Arc::clone(&raft_multi_raft),
        Arc::clone(&routing),
        Arc::clone(&topology),
        Arc::clone(&transport),
    ));

    let mut registry = SubsystemRegistry::new();
    register_default_subsystems(&mut registry, config, &ctx, executor)?;

    registry
        .start_all(&ctx)
        .await
        .map_err(|e| crate::error::ClusterError::Storage {
            detail: format!("subsystem start failed: {e}"),
        })
}