Function phase1_base_copy

nodedb-cluster/src/migration_executor/phases.rs:16–94

(
    ex: &MigrationExecutor,
    state: &mut MigrationState,
    group_id: u64,
    req: &MigrationRequest,
    migration_id: MigrationId,
)

use super::executor::{MigrationExecutor, MigrationRequest};

pub(super) async fn phase1_base_copy(
    ex: &MigrationExecutor,
    state: &mut MigrationState,
    group_id: u64,
    req: &MigrationRequest,
    migration_id: MigrationId,
) -> Result<()> {
    let committed = {
        let mr = ex.multi_raft.lock().unwrap_or_else(|p| p.into_inner());
        mr.group_statuses()
            .iter()
            .find(|s| s.group_id == group_id)
            .map(|s| s.commit_index)
            .unwrap_or(0)
    };
    state.start_base_copy(committed);

    ex.propose_checkpoint(
        migration_id,
        0,
        MigrationCheckpointPayload::AddLearner {
            vshard_id: req.vshard_id,
            source_node: req.source_node,
            target_node: req.target_node,
            source_group: group_id,
            write_pause_budget_us: req.write_pause_budget_us,
            started_at_hlc: nodedb_types::Hlc::default(),
        },
    )
    .await?;

    info!(
        vshard = req.vshard_id,
        group = group_id,
        target = req.target_node,
        entries = committed,
        "phase 1: adding target to raft group"
    );

    let change = ConfChange {
        change_type: ConfChangeType::AddLearner,
        node_id: req.target_node,
    };
    let learner_log_index = {
        let mut mr = ex.multi_raft.lock().unwrap_or_else(|p| p.into_inner());
        mr.propose_conf_change(group_id, &change)?;
        mr.group_statuses()
            .iter()
            .find(|s| s.group_id == group_id)
            .map(|s| s.commit_index)
            .unwrap_or(committed)
    };

    if let Some(node_info) = {
        let topo = ex.topology.read().unwrap_or_else(|p| p.into_inner());
        topo.get_node(req.target_node).map(|n| n.addr.clone())
    } && let Ok(addr) = node_info.parse()
    {