feat(heal): stage conservative legacy MRF migration evidence

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -33,6 +33,9 @@ use std::collections::HashMap;
 use tokio::io::AsyncReadExt;
 use uuid::Uuid;
 
+/// Explicit pending migration; never activates the production writer or GC.
+pub mod migration;
+
 // Root-level control files avoid requiring a new directory before the first
 // atomic commit. They remain inside the storage owner's metadata volume.
 const PAYLOAD_PATHS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
@@ -66,6 +69,23 @@ struct Manifest {
 }
 
 impl Manifest {
+    fn encode(owner: Uuid, sequence: u64, payload: &[u8]) -> Result<Vec<u8>, SnapshotError> {
+        let mut bytes = Vec::with_capacity(MANIFEST_LEN);
+        bytes.extend_from_slice(MAGIC);
+        bytes.push(VERSION);
+        bytes.extend_from_slice(owner.as_bytes());
+        bytes.extend_from_slice(&sequence.to_le_bytes());
+        bytes.extend_from_slice(
+            &u64::try_from(payload.len())
+                .map_err(|_| SnapshotError::TooLarge)?
+                .to_le_bytes(),
+        );
+        bytes.extend_from_slice(&Sha256::digest(payload));
+        bytes.extend_from_slice(&Sha256::digest(&bytes));
+        Self::decode(&bytes, payload.len())?;
+        Ok(bytes)
+    }
+
     fn decode(bytes: &[u8], limit: usize) -> Result<Self, SnapshotError> {
         if bytes.len() != MANIFEST_LEN || &bytes[..8] != MAGIC {
             return Err(SnapshotError::Corrupt);

crates/heal/src/heal/mrf_queue/snapshot/migration.rs

@@ -0,0 +1,900 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! Pending, owner-local legacy import. These paths are deliberately invisible
+//! to the active snapshot reader and legacy consumer. Source revalidation is
+//! not a writer freeze: no result here grants activation or reclamation rights.
+//! All source bytes and inherited responsibilities survive admission/replay.
+
+use super::{MANIFEST_LEN, Manifest, SnapshotError, read_bounded};
+use crate::heal::RUSTFS_META_BUCKET;
+use crate::heal::mrf_queue::{MRF_JOURNAL_PATH, MRF_SCOPED_JOURNAL_PATH, decode_one};
+use crate::heal::storage_api::owner::{
+    EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskStore,
+};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::collections::BTreeSet;
+use tokio::io::AsyncReadExt;
+use uuid::Uuid;
+
+const PAYLOADS: [&str; 2] = [".heal-mrf-import-pending.0.bin", ".heal-mrf-import-pending.1.bin"];
+const COMMITS: [&str; 2] = [".heal-mrf-import-commit.0.bin", ".heal-mrf-import-commit.1.bin"];
+const MAX_DISKS: usize = 64;
+const CLAIM: &str = ".heal-mrf-import-claim.bin";
+
+/// Limits apply to the complete encoded candidate and all distinct raw records,
+/// including inherited sources. Exceeding either preserves previous anchors.
+#[derive(Clone, Copy, Debug)]
+pub struct MigrationLimits {
+    pub max_bytes: usize,
+    pub max_records: usize,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum MigrationError {
+    #[error(transparent)]
+    Snapshot(#[from] SnapshotError),
+    #[error("MRF migration requires every configured, formatted local disk")]
+    CoverageGap,
+    #[error("MRF migration source changed; all recovery anchors are retained")]
+    SourceChanged,
+    #[error("MRF migration candidate is invalid")]
+    Invalid,
+    #[error("MRF migration has no responsibility evidence")]
+    Empty,
+    #[error("MRF migration conditional publication conflicted")]
+    Conflict,
+    #[error("MRF migration staging is claimed; interrupted claims require separately fenced recovery")]
+    Claimed,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+enum LegacyPath {
+    Scoped,
+    Mirror,
+}
+
+impl LegacyPath {
+    fn path(self) -> &'static str {
+        match self {
+            Self::Scoped => MRF_SCOPED_JOURNAL_PATH,
+            Self::Mirror => MRF_JOURNAL_PATH,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Source {
+    disk_id: Uuid,
+    path: LegacyPath,
+    digest: [u8; 32],
+    // None proves an observed absent path, distinct from a present empty file.
+    bytes: Option<Vec<u8>>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PendingMigration {
+    version: u8,
+    sources: Vec<Source>,
+    inherited: Vec<Source>,
+}
+
+impl PendingMigration {
+    /// Raw, complete records. No scope/version normalization, attempts pruning,
+    /// incarnation inference or task-success interpretation is performed.
+    pub fn replay_records(&self, limits: MigrationLimits) -> Result<Vec<Vec<u8>>, MigrationError> {
+        self.validate_limits(limits)?;
+        let mut records = BTreeSet::new();
+        for source in self.sources.iter().chain(&self.inherited) {
+            if source.disk_id.is_nil() {
+                return Err(MigrationError::Invalid);
+            }
+            let bytes = source.bytes.as_deref().unwrap_or_default();
+            if <[u8; 32]>::from(Sha256::digest(bytes)) != source.digest {
+                return Err(MigrationError::Invalid);
+            }
+            let mut offset = 0;
+            while offset < bytes.len() {
+                let (_, consumed) = decode_one(&bytes[offset..]).ok_or(MigrationError::Invalid)?;
+                let end = offset.checked_add(consumed).ok_or(MigrationError::Invalid)?;
+                records.insert(bytes[offset..end].to_vec());
+                if records.len() > limits.max_records {
+                    return Err(SnapshotError::TooLarge.into());
+                }
+                offset = end;
+            }
+        }
+        if records.is_empty() {
+            return Err(MigrationError::Empty);
+        }
+        Ok(records.into_iter().collect())
+    }
+
+    fn validate_limits(&self, limits: MigrationLimits) -> Result<(), MigrationError> {
+        if self.version != 1 {
+            return Err(SnapshotError::Unsupported.into());
+        }
+        if self.sources.is_empty() || self.sources.len() > MAX_DISKS * 2 {
+            return Err(MigrationError::Invalid);
+        }
+        // Bound the raw input before allocating the JSON representation.
+        let total = self
+            .sources
+            .iter()
+            .chain(&self.inherited)
+            .try_fold(0usize, |total, source| {
+                total
+                    .checked_add(source.bytes.as_ref().map_or(0, Vec::len))
+                    .and_then(|n| n.checked_add(128))
+            })
+            .ok_or(SnapshotError::TooLarge)?;
+        if total > limits.max_bytes {
+            return Err(SnapshotError::TooLarge.into());
+        }
+        Ok(())
+    }
+
+    fn encode(&self, limits: MigrationLimits) -> Result<Vec<u8>, MigrationError> {
+        self.replay_records(limits)?;
+        let bytes = serde_json::to_vec(self).map_err(|_| MigrationError::Invalid)?;
+        if bytes.len() > limits.max_bytes {
+            return Err(SnapshotError::TooLarge.into());
+        }
+        Ok(bytes)
+    }
+
+    async fn revalidate(&self, disks: &[EcstoreDiskStore], limits: MigrationLimits) -> Result<(), MigrationError> {
+        let current = capture(disks, limits).await?;
+        if current.sources != self.sources {
+            return Err(MigrationError::SourceChanged);
+        }
+        Ok(())
+    }
+}
+
+async fn configured_disks(disks: &[Option<EcstoreDiskStore>]) -> Result<Vec<EcstoreDiskStore>, MigrationError> {
+    if disks.is_empty() || disks.len() > MAX_DISKS {
+        return Err(MigrationError::CoverageGap);
+    }
+    let mut ordered = std::collections::BTreeMap::new();
+    for disk in disks {
+        let disk = disk.as_ref().ok_or(MigrationError::CoverageGap)?;
+        if !EcstoreDiskAPI::is_local(disk.as_ref()) {
+            return Err(MigrationError::CoverageGap);
+        }
+        let id = EcstoreDiskAPI::get_disk_id(disk.as_ref())
+            .await
+            .map_err(SnapshotError::Disk)?
+            .filter(|id| !id.is_nil())
+            .ok_or(MigrationError::CoverageGap)?;
+        if ordered.insert(id, disk.clone()).is_some() {
+            return Err(MigrationError::CoverageGap);
+        }
+    }
+    Ok(ordered.into_values().collect())
+}
+
+async fn capture(disks: &[EcstoreDiskStore], limits: MigrationLimits) -> Result<PendingMigration, MigrationError> {
+    let mut sources = Vec::with_capacity(disks.len() * 2);
+    let mut identities = BTreeSet::new();
+    let mut remaining = limits.max_bytes;
+    for disk in disks {
+        let id = EcstoreDiskAPI::get_disk_id(disk.as_ref())
+            .await
+            .map_err(SnapshotError::Disk)?
+            .filter(|id| !id.is_nil())
+            .ok_or(MigrationError::CoverageGap)?;
+        if !identities.insert(id) {
+            return Err(MigrationError::CoverageGap);
+        }
+        for path in [LegacyPath::Scoped, LegacyPath::Mirror] {
+            // A missing metadata volume is a coverage gap, not an absent journal.
+            let bytes = match EcstoreDiskAPI::read_file(disk.as_ref(), RUSTFS_META_BUCKET, path.path()).await {
+                Ok(reader) => {
+                    let maximum = u64::try_from(remaining.checked_add(1).ok_or(SnapshotError::TooLarge)?)
+                        .map_err(|_| SnapshotError::TooLarge)?;
+                    let mut bytes = Vec::new();
+                    reader
+                        .take(maximum)
+                        .read_to_end(&mut bytes)
+                        .await
+                        .map_err(SnapshotError::Read)?;
+                    if bytes.len() > remaining {
+                        return Err(SnapshotError::TooLarge.into());
+                    }
+                    Some(bytes)
+                }
+                Err(EcstoreDiskError::FileNotFound) => None,
+                Err(error) => return Err(SnapshotError::Disk(error).into()),
+            };
+            remaining = remaining
+                .checked_sub(bytes.as_ref().map_or(0, Vec::len))
+                .ok_or(SnapshotError::TooLarge)?;
+            let digest = Sha256::digest(bytes.as_deref().unwrap_or_default()).into();
+            sources.push(Source {
+                disk_id: id,
+                path,
+                digest,
+                bytes,
+            });
+        }
+    }
+    sources.sort_by_key(|source| (source.disk_id, matches!(source.path, LegacyPath::Mirror)));
+    let candidate = PendingMigration {
+        version: 1,
+        sources,
+        inherited: Vec::new(),
+    };
+    candidate.encode(limits)?;
+    Ok(candidate)
+}
+
+/// Inspect every configured source without merging v1 mirrors into a claimed
+/// latest snapshot. A complete subset/superset is retained as pending evidence.
+pub async fn capture_legacy_migration(
+    disks: &[Option<EcstoreDiskStore>],
+    limits: MigrationLimits,
+) -> Result<PendingMigration, MigrationError> {
+    capture(&configured_disks(disks).await?, limits).await
+}
+
+struct Staged {
+    manifest: Manifest,
+    candidate: PendingMigration,
+    slot: usize,
+}
+
+async fn read_staged(disks: &[EcstoreDiskStore], limits: MigrationLimits) -> Result<Option<Staged>, MigrationError> {
+    let mut selected: Option<Staged> = None;
+    let mut damage = None;
+    let mut identities = std::collections::BTreeMap::new();
+    for disk in disks {
+        for slot in 0..2 {
+            let result = async {
+                let Some(bytes) = read_bounded(disk, COMMITS[slot], MANIFEST_LEN).await? else {
+                    return Ok(None);
+                };
+                let manifest = Manifest::decode(&bytes, limits.max_bytes)?;
+                let payload = read_bounded(disk, PAYLOADS[slot], manifest.payload_len)
+                    .await?
+                    .ok_or(MigrationError::Invalid)?;
+                if payload.len() != manifest.payload_len || <[u8; 32]>::from(Sha256::digest(&payload)) != manifest.payload_digest
+                {
+                    return Err(MigrationError::Invalid);
+                }
+                let candidate: PendingMigration = serde_json::from_slice(&payload).map_err(|_| MigrationError::Invalid)?;
+                if candidate.encode(limits)? != payload {
+                    return Err(MigrationError::Invalid);
+                }
+                Ok(Some(Staged {
+                    manifest,
+                    candidate,
+                    slot,
+                }))
+            }
+            .await;
+            match result {
+                Ok(Some(next)) => {
+                    let identity = (next.manifest.owner, next.manifest.payload_digest);
+                    if identities
+                        .insert(next.manifest.sequence, identity)
+                        .is_some_and(|old| old != identity)
+                    {
+                        return Err(MigrationError::Conflict);
+                    }
+                    if selected
+                        .as_ref()
+                        .is_none_or(|old| old.manifest.sequence < next.manifest.sequence)
+                    {
+                        selected = Some(next);
+                    }
+                }
+                Ok(None) => {}
+                Err(MigrationError::Snapshot(SnapshotError::Unsupported)) => return Err(SnapshotError::Unsupported.into()),
+                Err(error) => damage = Some(error),
+            }
+        }
+    }
+    match (selected, damage) {
+        (Some(staged), _) => Ok(Some(staged)),
+        (None, Some(error)) => Err(error),
+        (None, None) => Ok(None),
+    }
+}
+
+async fn install(disk: &EcstoreDiskStore, path: &str, bytes: &[u8], limit: usize) -> Result<(), MigrationError> {
+    let expected = read_bounded(disk, path, limit).await?.map(EcstoreDiskBytes::from);
+    let result = EcstoreDiskAPI::compare_and_update_file(
+        disk.as_ref(),
+        RUSTFS_META_BUCKET,
+        path,
+        expected,
+        Some(EcstoreDiskBytes::copy_from_slice(bytes)),
+    )
+    .await
+    .map_err(SnapshotError::Disk)?;
+    if result != EcstoreConditionalFileUpdate::Updated {
+        return Err(MigrationError::Conflict);
+    }
+    Ok(())
+}
+
+/// Persist an explicitly requested pending import using the storage owner's CAS
+/// (and its configured metadata durability). This does not freeze legacy ingress
+/// or grant a durable-acceptance/GC receipt. Activation requires W14/W21 evidence.
+pub async fn stage_legacy_migration(
+    disks: &[Option<EcstoreDiskStore>],
+    candidate: &PendingMigration,
+    owner: Uuid,
+    limits: MigrationLimits,
+) -> Result<u64, MigrationError> {
+    let disks = configured_disks(disks).await?;
+    // Claim every configured disk in identity order. A crash/cancellation leaves
+    // claims intact; a new process cannot guess that the old writer is fenced.
+    let claim = EcstoreDiskBytes::copy_from_slice(Uuid::new_v4().as_bytes());
+    let mut claimed = Vec::new();
+    let result = async {
+        for disk in &disks {
+            match EcstoreDiskAPI::compare_and_update_file(disk.as_ref(), RUSTFS_META_BUCKET, CLAIM, None, Some(claim.clone()))
+                .await
+                .map_err(SnapshotError::Disk)?
+            {
+                EcstoreConditionalFileUpdate::Updated => claimed.push(disk.clone()),
+                _ => return Err(MigrationError::Claimed),
+            }
+        }
+        stage_claimed(&disks, candidate, owner, limits).await
+    }
+    .await;
+    for disk in claimed {
+        let released =
+            EcstoreDiskAPI::compare_and_update_file(disk.as_ref(), RUSTFS_META_BUCKET, CLAIM, Some(claim.clone()), None)
+                .await
+                .map_err(SnapshotError::Disk)?;
+        if released != EcstoreConditionalFileUpdate::Updated {
+            return Err(MigrationError::Claimed);
+        }
+    }
+    result
+}
+
+async fn stage_claimed(
+    disks: &[EcstoreDiskStore],
+    candidate: &PendingMigration,
+    owner: Uuid,
+    limits: MigrationLimits,
+) -> Result<u64, MigrationError> {
+    candidate.revalidate(&disks, limits).await?;
+    let previous = read_staged(&disks, limits).await?;
+    if previous.as_ref().is_some_and(|old| old.manifest.owner != owner) {
+        return Err(MigrationError::Conflict);
+    }
+    let mut candidate = candidate.clone();
+    if let Some(old) = &previous {
+        for source in old.candidate.sources.iter().chain(&old.candidate.inherited) {
+            if !candidate.sources.contains(source) && !candidate.inherited.contains(source) {
+                candidate.inherited.push(source.clone());
+            }
+        }
+    }
+    let payload = candidate.encode(limits)?;
+    let digest: [u8; 32] = Sha256::digest(&payload).into();
+    if let Some(old) = &previous
+        && old.manifest.payload_digest == digest
+    {
+        candidate.revalidate(&disks, limits).await?;
+        return Ok(old.manifest.sequence);
+    }
+    let sequence = previous.as_ref().map_or(1, |old| old.manifest.sequence.saturating_add(1));
+    let slot = previous.as_ref().map_or(0, |old| 1 - old.slot);
+    let manifest = Manifest::encode(owner, sequence, &payload)?;
+    for disk in disks {
+        install(disk, PAYLOADS[slot], &payload, limits.max_bytes).await?;
+    }
+    #[cfg(test)]
+    tests::interrupt_at(owner, tests::Boundary::AfterPayload).await?;
+    candidate.revalidate(&disks, limits).await?;
+    #[cfg(test)]
+    tests::interrupt_at(owner, tests::Boundary::BeforeManifest).await?;
+    for disk in disks {
+        install(disk, COMMITS[slot], &manifest, MANIFEST_LEN).await?;
+    }
+    #[cfg(test)]
+    tests::interrupt_at(owner, tests::Boundary::AfterManifest).await?;
+    let recovered = read_staged(&disks, limits).await?.ok_or(MigrationError::Invalid)?;
+    if recovered.manifest.sequence != sequence || recovered.manifest.payload_digest != digest {
+        return Err(MigrationError::Conflict);
+    }
+    recovered.candidate.revalidate(&disks, limits).await?;
+    #[cfg(test)]
+    tests::interrupt_at(owner, tests::Boundary::AfterReadback).await?;
+    Ok(sequence)
+}
+
+/// Reload pending obligations after process restart. Manager admission never
+/// removes them. Missing/changed sources block migration, preserving all files.
+pub async fn recover_pending_migration(
+    disks: &[Option<EcstoreDiskStore>],
+    limits: MigrationLimits,
+) -> Result<Option<PendingMigration>, MigrationError> {
+    let disks = configured_disks(disks).await?;
+    let Some(staged) = read_staged(&disks, limits).await? else {
+        return Ok(None);
+    };
+    staged.candidate.revalidate(&disks, limits).await?;
+    Ok(Some(staged.candidate))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::heal::mrf_queue::encode_intent;
+    use crate::heal::{DiskOption, Endpoint, new_disk};
+    use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfScope};
+    use std::sync::Arc;
+    use tempfile::TempDir;
+
+    const LIMITS: MigrationLimits = MigrationLimits {
+        max_bytes: 64 * 1024,
+        max_records: 100,
+    };
+
+    #[derive(Clone, Copy, PartialEq, Eq)]
+    pub(super) enum Boundary {
+        AfterPayload,
+        BeforeManifest,
+        AfterManifest,
+        AfterReadback,
+    }
+
+    static INTERRUPTIONS: std::sync::LazyLock<std::sync::Mutex<std::collections::BTreeMap<Uuid, Boundary>>> =
+        std::sync::LazyLock::new(Default::default);
+
+    static SOURCE_CHANGES: std::sync::LazyLock<std::sync::Mutex<std::collections::BTreeMap<Uuid, (EcstoreDiskStore, Vec<u8>)>>> =
+        std::sync::LazyLock::new(Default::default);
+
+    pub(super) async fn interrupt_at(owner: Uuid, boundary: Boundary) -> Result<(), MigrationError> {
+        if boundary == Boundary::AfterPayload {
+            let change = SOURCE_CHANGES.lock().expect("source fault map").remove(&owner);
+            if let Some((disk, bytes)) = change {
+                source(&disk, &bytes).await;
+            }
+        }
+        let mut interruptions = INTERRUPTIONS.lock().expect("fault map");
+        if interruptions.get(&owner) == Some(&boundary) {
+            interruptions.remove(&owner);
+            return Err(SnapshotError::Read(std::io::Error::new(
+                std::io::ErrorKind::Interrupted,
+                "injected migration boundary failure",
+            ))
+            .into());
+        }
+        Ok(())
+    }
+
+    fn record(object: &str, kind: MrfKind, scope: Option<MrfScope>) -> Vec<u8> {
+        let mut bytes = Vec::new();
+        assert!(encode_intent(
+            &MrfIntent {
+                bucket: Arc::from("bucket"),
+                object: Arc::from(object),
+                version_id: None,
+                kind,
+                scope,
+                lease: None,
+                enqueued_at_ms: 1,
+                attempts: 255,
+            },
+            &mut bytes
+        ));
+        bytes
+    }
+
+    async fn disk(root: &TempDir, name: &str) -> EcstoreDiskStore {
+        let path = root.path().join(name);
+        std::fs::create_dir_all(&path).expect("create test disk");
+        let mut endpoint = Endpoint::try_from(path.to_string_lossy().as_ref()).expect("disk endpoint");
+        endpoint.set_idx = 0;
+        endpoint.disk_idx = 0;
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("open disk");
+        let created = EcstoreDiskAPI::make_volume(disk.as_ref(), RUSTFS_META_BUCKET).await;
+        assert!(
+            matches!(created, Ok(()) | Err(EcstoreDiskError::VolumeExists)),
+            "metadata volume: {created:?}"
+        );
+        let id = Uuid::new_v4();
+        let format = serde_json::json!({
+            "version": "1", "format": "xl-single", "id": Uuid::new_v4(),
+            "xl": { "version": "3", "this": id, "sets": [[id]], "distributionAlgo": "SIPMOD+PARITY" }
+        });
+        EcstoreDiskAPI::write_all(
+            disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            "format.json",
+            serde_json::to_vec(&format).expect("format").into(),
+        )
+        .await
+        .expect("format disk");
+        assert_eq!(EcstoreDiskAPI::get_disk_id(disk.as_ref()).await.expect("formatted identity"), Some(id));
+        disk
+    }
+
+    async fn source(disk: &EcstoreDiskStore, bytes: &[u8]) {
+        // Legacy source paths predate the root-level COW control files.
+        EcstoreDiskAPI::write_all(
+            disk.as_ref(),
+            RUSTFS_META_BUCKET,
+            MRF_SCOPED_JOURNAL_PATH,
+            EcstoreDiskBytes::copy_from_slice(bytes),
+        )
+        .await
+        .expect("write legacy source");
+    }
+
+    #[tokio::test]
+    async fn migration_subset_union_preserves_sources_across_reopen() {
+        let root = TempDir::new().expect("test directory");
+        let first = disk(&root, "first").await;
+        let second = disk(&root, "second").await;
+        let a = record("a", MrfKind::PartialWrite, None);
+        let b = record(
+            "b",
+            MrfKind::DecodeFailure,
+            Some(MrfScope {
+                pool_index: 0,
+                set_index: 1,
+            }),
+        );
+        source(&first, &a).await;
+        source(&second, &[a.clone(), b.clone()].concat()).await;
+        let disks = [Some(first.clone()), Some(second.clone())];
+        let candidate = capture_legacy_migration(&disks, LIMITS)
+            .await
+            .expect("capture both complete sources");
+        let reverse = capture_legacy_migration(&[Some(second.clone()), Some(first.clone())], LIMITS)
+            .await
+            .expect("reverse disk order");
+        assert_eq!(
+            candidate.encode(LIMITS).expect("candidate"),
+            reverse.encode(LIMITS).expect("reversed candidate")
+        );
+        assert_eq!(candidate.replay_records(LIMITS).expect("raw records").len(), 2);
+        let owner = Uuid::new_v4();
+        assert_eq!(
+            stage_legacy_migration(&disks, &candidate, owner, LIMITS)
+                .await
+                .expect("stage candidate"),
+            1
+        );
+        drop(candidate);
+        let recovered = recover_pending_migration(&disks, LIMITS)
+            .await
+            .expect("restart read")
+            .expect("pending import");
+        // Reading or discarding a replay batch must not consume its stored anchor.
+        let mut admitted = recovered.replay_records(LIMITS).expect("replay batch");
+        admitted.pop();
+        drop(admitted);
+        assert_eq!(
+            recover_pending_migration(&disks, LIMITS)
+                .await
+                .expect("second restart")
+                .expect("anchor")
+                .replay_records(LIMITS)
+                .expect("records")
+                .len(),
+            2
+        );
+        assert_eq!(
+            stage_legacy_migration(&disks, &recovered, owner, LIMITS)
+                .await
+                .expect("idempotent retry"),
+            1
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(first.as_ref(), RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                .await
+                .expect("first source"),
+            a
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(second.as_ref(), RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                .await
+                .expect("second source"),
+            [a, b].concat()
+        );
+        assert!(
+            super::super::read_committed(&[first, second], LIMITS.max_bytes)
+                .await
+                .expect("active reader")
+                .is_none(),
+            "pending import must not activate the production snapshot"
+        );
+    }
+
+    #[tokio::test]
+    async fn migration_source_change_and_capacity_failure_keep_old_commit() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let disks = [Some(disk.clone())];
+        let owner = Uuid::new_v4();
+        let a = record("a", MrfKind::PartialWrite, None);
+        let b = record("b", MrfKind::PartialWrite, None);
+        source(&disk, &a).await;
+        let original = capture_legacy_migration(&disks, LIMITS).await.expect("initial capture");
+        stage_legacy_migration(&disks, &original, owner, LIMITS)
+            .await
+            .expect("initial commit");
+        let before = EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, COMMITS[0])
+            .await
+            .expect("old commit");
+        source(&disk, &b).await;
+        assert!(matches!(
+            stage_legacy_migration(&disks, &original, owner, LIMITS).await,
+            Err(MigrationError::SourceChanged)
+        ));
+        let next = capture_legacy_migration(&disks, LIMITS).await.expect("new source capture");
+        assert!(matches!(
+            stage_legacy_migration(
+                &disks,
+                &next,
+                owner,
+                MigrationLimits {
+                    max_records: 1,
+                    ..LIMITS
+                }
+            )
+            .await,
+            Err(MigrationError::Snapshot(SnapshotError::TooLarge))
+        ));
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, COMMITS[0])
+                .await
+                .expect("retained commit"),
+            before
+        );
+        assert_eq!(
+            stage_legacy_migration(&disks, &next, owner, LIMITS)
+                .await
+                .expect("COW successor"),
+            2
+        );
+        let records = recover_pending_migration(&disks, LIMITS)
+            .await
+            .expect("successor restart")
+            .expect("successor")
+            .replay_records(LIMITS)
+            .expect("responsibilities");
+        assert!(
+            records.contains(&a) && records.contains(&b),
+            "successor must inherit old source responsibilities"
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, COMMITS[0])
+                .await
+                .expect("previous slot retained"),
+            before
+        );
+    }
+
+    #[tokio::test]
+    async fn migration_torn_inactive_payload_and_manifest_keep_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let disks = [Some(disk.clone())];
+        source(&disk, &record("a", MrfKind::PartialWrite, None)).await;
+        let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture");
+        stage_legacy_migration(&disks, &candidate, Uuid::new_v4(), LIMITS)
+            .await
+            .expect("initial commit");
+        for (path, bytes) in [
+            (PAYLOADS[1], b"torn payload".as_slice()),
+            (COMMITS[1], b"torn manifest".as_slice()),
+        ] {
+            install(&disk, path, bytes, LIMITS.max_bytes)
+                .await
+                .expect("interrupted inactive write");
+            assert_eq!(
+                recover_pending_migration(&disks, LIMITS)
+                    .await
+                    .expect("recover previous")
+                    .expect("anchor")
+                    .replay_records(LIMITS)
+                    .expect("retained responsibility")
+                    .len(),
+                1
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn migration_missing_corrupt_and_empty_sources_fail_closed() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        assert!(matches!(
+            capture_legacy_migration(&[Some(disk.clone()), None], LIMITS).await,
+            Err(MigrationError::CoverageGap)
+        ));
+        assert!(matches!(
+            capture_legacy_migration(&[Some(disk.clone())], LIMITS).await,
+            Err(MigrationError::Empty)
+        ));
+        source(&disk, b"corrupt").await;
+        assert!(matches!(
+            capture_legacy_migration(&[Some(disk)], LIMITS).await,
+            Err(MigrationError::Invalid)
+        ));
+    }
+
+    #[tokio::test]
+    async fn migration_commit_boundaries_and_lost_response_recover_idempotently() {
+        for boundary in [
+            Boundary::AfterPayload,
+            Boundary::BeforeManifest,
+            Boundary::AfterManifest,
+            Boundary::AfterReadback,
+        ] {
+            let root = TempDir::new().expect("test directory");
+            let disk = disk(&root, "disk").await;
+            let disks = [Some(disk.clone())];
+            let bytes = record("a", MrfKind::PartialWrite, None);
+            source(&disk, &bytes).await;
+            let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture");
+            let owner = Uuid::new_v4();
+            INTERRUPTIONS.lock().expect("fault map").insert(owner, boundary);
+            assert!(stage_legacy_migration(&disks, &candidate, owner, LIMITS).await.is_err());
+            assert_eq!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                    .await
+                    .expect("source survives interruption"),
+                bytes
+            );
+            let recovery = recover_pending_migration(&disks, LIMITS).await.expect("restart inspection");
+            assert_eq!(recovery.is_some(), matches!(boundary, Boundary::AfterManifest | Boundary::AfterReadback));
+            assert_eq!(
+                stage_legacy_migration(&disks, &candidate, owner, LIMITS)
+                    .await
+                    .expect("retry interrupted stage"),
+                1
+            );
+            assert_eq!(
+                recover_pending_migration(&disks, LIMITS)
+                    .await
+                    .expect("restart after retry")
+                    .expect("anchor")
+                    .replay_records(LIMITS)
+                    .expect("records"),
+                vec![bytes]
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn migration_interrupted_claim_does_not_authorize_takeover() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let disks = [Some(disk.clone())];
+        source(&disk, &record("a", MrfKind::PartialWrite, None)).await;
+        let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture");
+        let owner = Uuid::new_v4();
+        stage_legacy_migration(&disks, &candidate, owner, LIMITS)
+            .await
+            .expect("committed anchor");
+        install(&disk, CLAIM, b"interrupted writer", 64)
+            .await
+            .expect("interrupted claim");
+        assert!(matches!(
+            stage_legacy_migration(&disks, &candidate, owner, LIMITS).await,
+            Err(MigrationError::Claimed)
+        ));
+        assert_eq!(
+            recover_pending_migration(&disks, LIMITS)
+                .await
+                .expect("recovery remains read-only")
+                .expect("anchor")
+                .replay_records(LIMITS)
+                .expect("record")
+                .len(),
+            1
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, CLAIM)
+                .await
+                .expect("claim retained"),
+            b"interrupted writer".as_slice()
+        );
+    }
+
+    #[tokio::test]
+    async fn migration_source_change_after_payload_prevents_manifest_publication() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let disks = [Some(disk.clone())];
+        source(&disk, &record("a", MrfKind::PartialWrite, None)).await;
+        let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture");
+        let owner = Uuid::new_v4();
+        let changed = record("b", MrfKind::PartialWrite, None);
+        SOURCE_CHANGES
+            .lock()
+            .expect("source fault map")
+            .insert(owner, (disk.clone(), changed.clone()));
+        assert!(matches!(
+            stage_legacy_migration(&disks, &candidate, owner, LIMITS).await,
+            Err(MigrationError::SourceChanged)
+        ));
+        assert!(
+            recover_pending_migration(&disks, LIMITS)
+                .await
+                .expect("no committed import")
+                .is_none()
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                .await
+                .expect("changed source survives"),
+            changed
+        );
+        assert!(
+            read_bounded(&disk, PAYLOADS[0], LIMITS.max_bytes)
+                .await
+                .expect("candidate retained")
+                .is_some()
+        );
+    }
+
+    #[test]
+    fn migration_raw_identity_preserves_kind_scope_and_nil_version() {
+        let id = Uuid::new_v4();
+        let mut variants = Vec::new();
+        for kind in [MrfKind::PartialWrite, MrfKind::DecodeFailure] {
+            for scope in [
+                None,
+                Some(MrfScope {
+                    pool_index: 0,
+                    set_index: 0,
+                }),
+                Some(MrfScope {
+                    pool_index: 0,
+                    set_index: 1,
+                }),
+            ] {
+                variants.push(record("same", kind, scope));
+            }
+        }
+        let mut nil = record("same", MrfKind::PartialWrite, None);
+        nil[12] = 1;
+        nil.splice(13..13, [0; 16]);
+        let end = nil.len() - 4;
+        let mut crc = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        crc.update(&nil[..end]);
+        nil[end..].copy_from_slice(&u32::try_from(crc.finalize()).expect("CRC").to_le_bytes());
+        variants.push(nil);
+        let bytes = variants.concat();
+        let candidate = PendingMigration {
+            version: 1,
+            sources: vec![Source {
+                disk_id: id,
+                path: LegacyPath::Scoped,
+                digest: Sha256::digest(&bytes).into(),
+                bytes: Some(bytes),
+            }],
+            inherited: Vec::new(),
+        };
+        let records = candidate.replay_records(LIMITS).expect("raw identities");
+        assert_eq!(records.len(), variants.len());
+        for variant in variants {
+            assert!(records.contains(&variant));
+        }
+    }
+}