Function apply_migration_checkpoint

nodedb-cluster/src/metadata_group/cache.rs:158–189

Apply a `MigrationCheckpoint` entry against the shared state table. Returns `Err` on CRC32C mismatch (fatal — corruption is louder than data loss) and on storage failures. Idempotent: if `(migration_id, phase, attempt)` is already present, the call is a no-op.

(
    table: &SharedMigrationStateTable,
    migration_id: uuid::Uuid,
    _phase: MigrationPhaseTag,
    attempt: u32,
    payload: crate::metadata_group::migration_state::MigrationCheckpointPayload,

/// loss) and on storage failures. Idempotent: if
/// `(migration_id, phase, attempt)` is already present, the call is a no-op.
pub fn apply_migration_checkpoint(
    table: &SharedMigrationStateTable,
    migration_id: uuid::Uuid,
    _phase: MigrationPhaseTag,
    attempt: u32,
    payload: crate::metadata_group::migration_state::MigrationCheckpointPayload,
    expected_crc: u32,
    ts_ms: u64,
) -> Result<(), ClusterError> {
    // Validate CRC32C against the encoded payload bytes.
    let actual_crc = payload.crc32c()?;
    if actual_crc != expected_crc {
        return Err(ClusterError::MigrationCheckpoint(
            MigrationCheckpointError::Crc32cMismatch {
                migration_id,
                expected: expected_crc,
                actual: actual_crc,
            },
        ));
    }

    let row = PersistedMigrationCheckpoint {
        migration_id: migration_id.hyphenated().to_string(),
        attempt,
        payload,
        crc32c: actual_crc,
        ts_ms,
    };

    let mut guard = table.lock().unwrap_or_else(|p| p.into_inner());
    guard.upsert(row)
}

/// Apply a `MigrationAbort` entry against the shared state table and
/// live routing table.