Function phase2_wal_catchup

nodedb-cluster/src/migration_executor/phases.rs:96–216

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

}

pub(super) async fn phase2_wal_catchup(
    ex: &MigrationExecutor,
    state: &mut MigrationState,
    group_id: u64,
    req: &MigrationRequest,
    migration_id: MigrationId,
) -> Result<()> {
    let leader_commit = {
        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_wal_catchup(leader_commit, leader_commit);

    info!(
        vshard = req.vshard_id,
        leader_commit, "phase 2: monitoring replication lag"
    );

    let initial_stable_id = ex.transport.peer_connection_stable_id(req.target_node);
    let initial_target_addr = {
        let topo = ex.topology.read().unwrap_or_else(|p| p.into_inner());
        topo.get_node(req.target_node).map(|n| n.addr.clone())
    };

    let poll_interval = Duration::from_millis(100);
    let timeout = Duration::from_secs(60);
    let deadline = std::time::Instant::now() + timeout;

    loop {
        tokio::time::sleep(poll_interval).await;

        if let Some(initial_id) = initial_stable_id {
            match ex.transport.peer_connection_stable_id(req.target_node) {
                Some(current_id) if current_id != initial_id => {
                    let reason = format!(
                        "peer identity changed mid-migration: stable_id {} -> {} for node {}",
                        initial_id, current_id, req.target_node
                    );
                    state.fail(reason.clone());
                    return Err(ClusterError::Transport { detail: reason });
                }
                None => {
                    let reason = format!(
                        "connection to target node {} lost during migration",
                        req.target_node
                    );
                    state.fail(reason.clone());
                    return Err(ClusterError::Transport { detail: reason });
                }
                _ => {}
            }
        }

        {