///|
fn migration_phase_rank(phase : MigrationPhase) -> Int {
match phase {
AddTarget => 0
Backfill => 1
Verify => 2
SwitchPrimary => 3
RemoveSource => 4
}
}
///|
fn action_touches_node(action : MigrationAction, node_id : String) -> Bool {
action.from_node == Some(node_id) || action.to_node == Some(node_id)
}
///|
fn node_pressure(actions : Array[MigrationAction], node_id : String) -> Int {
let mut count = 0
for action in actions {
if action_touches_node(action, node_id) {
count = count + 1
}
}
count
}
///|
fn exceeds_node_budget(
wave : Array[MigrationAction],
action : MigrationAction,
max_actions_per_node : Int,
) -> Bool {
match action.from_node {
Some(node_id) =>
if node_pressure(wave, node_id) >= max_actions_per_node {
return true
}
None => ()
}
match action.to_node {
Some(node_id) =>
if action.from_node != Some(node_id) &&
node_pressure(wave, node_id) >= max_actions_per_node {
return true
}
None => ()
}
false
}
///|
fn append_phase_waves(
waves : Array[MigrationWave],
phase : MigrationPhase,
actions : Array[MigrationAction],
max_actions_per_wave : Int,
max_actions_per_node : Int,
) -> Unit {
let current : Array[MigrationAction] = []
for action in actions {
if current.length() >= max_actions_per_wave ||
exceeds_node_budget(current, action, max_actions_per_node) {
waves.push({ index: waves.length(), phase, actions: current.copy() })
current.clear()
}
current.push(action)
}
if current.length() > 0 {
waves.push({ index: waves.length(), phase, actions: current.copy() })
}
}
///|
fn add_action(
phase_actions : Array[MigrationAction],
key : String,
phase : MigrationPhase,
from_node : String?,
to_node : String?,
) -> Unit {
let action = { key, phase, from_node, to_node }
phase_actions.push(action)
}
///|
fn append_actions(
target : Array[MigrationAction],
source : Array[MigrationAction],
) -> Unit {
for action in source {
target.push(action)
}
}
///|
/// Builds a five-phase, budgeted migration workflow.
///
/// All target replicas are added, backfilled, and verified before any primary
/// switch or source cleanup is emitted. Keys without a readable source are
/// reported as blocked instead of receiving unsafe actions.
pub fn plan_safe_migration(
keys : Array[String],
before : Array[ShardNode],
after : Array[ShardNode],
replicas? : Int = 1,
max_actions_per_wave? : Int = 64,
max_actions_per_node? : Int = 8,
salt? : String = "moonshardkit",
) -> SafeMigrationPlan {
let wave_budget = if max_actions_per_wave < 1 {
1
} else {
max_actions_per_wave
}
let node_budget = if max_actions_per_node < 1 {
1
} else {
max_actions_per_node
}
let all : Array[MigrationAction] = []
let additions : Array[MigrationAction] = []
let backfills : Array[MigrationAction] = []
let verifications : Array[MigrationAction] = []
let switches : Array[MigrationAction] = []
let removals : Array[MigrationAction] = []
let blocked_keys : Array[String] = []
let mut migrating_keys = 0
for key in keys {
let old_placement = place_replicas(before, key, replicas, salt~)
let new_placement = place_replicas(after, key, replicas, salt~)
if old_placement.owners == new_placement.owners {
continue
}
if old_placement.owners.length() == 0 && new_placement.owners.length() > 0 {
blocked_keys.push(key)
continue
}
migrating_keys = migrating_keys + 1
let source = first_owner(old_placement.owners)
for owner in new_placement.owners {
if !old_placement.owners.contains(owner) {
add_action(additions, key, AddTarget, source, Some(owner))
add_action(backfills, key, Backfill, source, Some(owner))
add_action(verifications, key, Verify, source, Some(owner))
}
}
let old_primary = first_owner(old_placement.owners)
let new_primary = first_owner(new_placement.owners)
if old_primary != new_primary {
add_action(switches, key, SwitchPrimary, old_primary, new_primary)
}
for owner in old_placement.owners {
if !new_placement.owners.contains(owner) {
add_action(removals, key, RemoveSource, Some(owner), None)
}
}
}
append_actions(all, additions)
append_actions(all, backfills)
append_actions(all, verifications)
append_actions(all, switches)
append_actions(all, removals)
let waves : Array[MigrationWave] = []
append_phase_waves(waves, AddTarget, additions, wave_budget, node_budget)
append_phase_waves(waves, Backfill, backfills, wave_budget, node_budget)
append_phase_waves(waves, Verify, verifications, wave_budget, node_budget)
append_phase_waves(waves, SwitchPrimary, switches, wave_budget, node_budget)
append_phase_waves(waves, RemoveSource, removals, wave_budget, node_budget)
{
keys: keys.length(),
migrating_keys,
blocked_keys,
actions: all,
waves,
max_actions_per_wave: wave_budget,
max_actions_per_node: node_budget,
}
}
///|
/// Checks phase ordering, wave indexes, and both concurrency budgets.
pub fn validate_safe_migration(plan : SafeMigrationPlan) -> Array[String] {
let issues : Array[String] = []
let mut previous_rank = -1
for index = 0; index < plan.waves.length(); index = index + 1 {
let wave = plan.waves[index]
let rank = migration_phase_rank(wave.phase)
if wave.index != index {
issues.push("wave index mismatch at \{index}")
}
if rank < previous_rank {
issues.push("phase order regressed at wave \{index}")
}
previous_rank = rank
if wave.actions.length() > plan.max_actions_per_wave {
issues.push("wave \{index} exceeds action budget")
}
for action in wave.actions {
if action.phase != wave.phase {
issues.push("wave \{index} mixes migration phases")
}
match action.from_node {
Some(node_id) =>
if node_pressure(wave.actions, node_id) > plan.max_actions_per_node {
issues.push("wave \{index} overloads node \{node_id}")
}
None => ()
}
match action.to_node {
Some(node_id) =>
if node_pressure(wave.actions, node_id) > plan.max_actions_per_node {
issues.push("wave \{index} overloads node \{node_id}")
}
None => ()
}
}
}
issues
}