Function phase3_cutover

nodedb-cluster/src/migration_executor/phases.rs:218–315

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

}

pub(super) async fn phase3_cutover(
    ex: &MigrationExecutor,
    state: &mut MigrationState,
    group_id: u64,
    req: &MigrationRequest,
    migration_id: MigrationId,
) -> Result<()> {
    let estimated_pause_us = 10_000;

    state.start_cutover(estimated_pause_us).map_err(|e| {
        state.fail(format!("cutover rejected: {e}"));
        e
    })?;

    let cutover_start = std::time::Instant::now();

    ex.propose_checkpoint(
        migration_id,
        3,
        MigrationCheckpointPayload::LeadershipTransfer {
            vshard_id: req.vshard_id,
            target_is_voter: true,
            new_leader_node_id: req.target_node,
            source_group: group_id,
        },
    )
    .await?;

    info!(
        vshard = req.vshard_id,
        estimated_pause_us, "phase 3: atomic cut-over"
    );

    if let Some(proposer) = &ex.metadata_proposer {
        let entry = Entry::RoutingChange(RoutingChange::LeadershipTransfer {
            group_id,
            new_leader_node_id: req.target_node,
        });
        proposer.propose_and_wait(entry).await?;
    } else {
        let mut routing = ex.routing.write().unwrap_or_else(|p| p.into_inner());
        routing.set_leader(group_id, req.target_node);
    }

    ex.propose_checkpoint(
        migration_id,
        4,
        MigrationCheckpointPayload::Cutover {
            vshard_id: req.vshard_id,
            new_leader_node_id: req.target_node,
            source_group: group_id,
        },
    )
    .await?;

    let ghost_stub = GhostStub {
        node_id: format!("vshard-{}", req.vshard_id),
        target_shard: req.vshard_id,