diff --git a/crates/heal/src/heal/mrf_queue/snapshot.rs b/crates/heal/src/heal/mrf_queue/snapshot.rs index e8d51ed9e..d39e0fafa 100644 --- a/crates/heal/src/heal/mrf_queue/snapshot.rs +++ b/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, 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 { if bytes.len() != MANIFEST_LEN || &bytes[..8] != MAGIC { return Err(SnapshotError::Corrupt); diff --git a/crates/heal/src/heal/mrf_queue/snapshot/migration.rs b/crates/heal/src/heal/mrf_queue/snapshot/migration.rs new file mode 100644 index 000000000..97dedaa59 --- /dev/null +++ b/crates/heal/src/heal/mrf_queue/snapshot/migration.rs @@ -0,0 +1,1450 @@ +// 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, + pub max_sources: 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, Hash, 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>, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PendingMigration { + version: u8, + sources: Vec, + inherited: Vec, +} + +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>, 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; + } + } + 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); + } + if self + .sources + .len() + .checked_add(self.inherited.len()) + .is_none_or(|count| count > limits.max_sources) + { + return Err(SnapshotError::TooLarge.into()); + } + // 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, 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]) -> Result, 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 { + 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], + limits: MigrationLimits, +) -> Result { + capture(&configured_disks(disks).await?, limits).await +} + +struct Staged { + manifest: Manifest, + candidate: PendingMigration, + slot: usize, +} + +type PayloadIdentity = (usize, [u8; 32]); + +struct StagingLineage { + latest: Option, + // Each collection has at most two identities per configured disk. + committed_payloads: BTreeSet, + orphaned_payloads: Vec, +} + +fn payload_identity(payload: &[u8]) -> PayloadIdentity { + (payload.len(), Sha256::digest(payload).into()) +} + +impl StagingLineage { + fn validate_orphans(&self, retry_payload: Option<&[u8]>) -> Result<(), MigrationError> { + let retry_identity = retry_payload.map(payload_identity); + if self + .orphaned_payloads + .iter() + .any(|identity| Some(*identity) != retry_identity && !self.committed_payloads.contains(identity)) + { + return Err(MigrationError::Conflict); + } + Ok(()) + } +} + +async fn read_staging_lineage(disks: &[EcstoreDiskStore], limits: MigrationLimits) -> Result { + let mut selected: Option = None; + let mut identities = std::collections::BTreeMap::new(); + let mut orphaned_payloads = Vec::new(); + let mut committed_payloads = BTreeSet::new(); + let mut mismatched_manifests = Vec::new(); + for disk in disks { + for slot in 0..2 { + let result = async { + let payload = read_bounded(disk, PAYLOADS[slot], limits.max_bytes).await?; + let Some(bytes) = read_bounded(disk, COMMITS[slot], MANIFEST_LEN).await? else { + if let Some(payload) = payload.as_deref() { + orphaned_payloads.push(payload_identity(payload)); + } + return Ok(None); + }; + let manifest = Manifest::decode(&bytes, limits.max_bytes)?; + let identity = (manifest.owner, manifest.payload_digest); + if identities + .insert(manifest.sequence, identity) + .is_some_and(|old| old != identity) + { + return Err(MigrationError::Conflict); + } + let payload = payload.ok_or(MigrationError::Invalid)?; + if payload.len() != manifest.payload_len || payload_identity(&payload).1 != manifest.payload_digest { + // A reused slot can hold the next retry payload while the + // old manifest still names the previous generation. + orphaned_payloads.push(payload_identity(&payload)); + mismatched_manifests.push(manifest); + return Ok(None); + } + let candidate: PendingMigration = serde_json::from_slice(&payload).map_err(|_| MigrationError::Invalid)?; + if candidate.encode(limits)? != payload || candidate.replay_records(limits)?.is_empty() { + return Err(MigrationError::Invalid); + } + Ok(Some(Staged { + manifest, + candidate, + slot, + })) + } + .await; + match result { + Ok(Some(next)) => { + committed_payloads.insert((next.manifest.payload_len, next.manifest.payload_digest)); + if selected + .as_ref() + .is_none_or(|old| old.manifest.sequence < next.manifest.sequence) + { + selected = Some(next); + } + } + Ok(None) => {} + Err(error) => return Err(error), + } + } + } + // Only a validated successor can supersede an unmatched older manifest. + // A newer or unordered manifest may still name responsibilities absent from + // the remaining payload, even if that payload matches another valid slot. + if mismatched_manifests.iter().any(|manifest| { + selected + .as_ref() + .is_none_or(|latest| latest.manifest.owner != manifest.owner || latest.manifest.sequence <= manifest.sequence) + }) { + return Err(MigrationError::Invalid); + } + Ok(StagingLineage { + latest: selected, + committed_payloads, + orphaned_payloads, + }) +} + +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], + candidate: &PendingMigration, + owner: Uuid, + limits: MigrationLimits, +) -> Result { + 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; + let mut release_error = None; + for disk in claimed { + let release = async { + #[cfg(test)] + tests::interrupt_at(owner, tests::Boundary::Release).await?; + 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); + } + Ok(()) + } + .await; + if let Err(error) = release + && release_error.is_none() + { + release_error = Some(error); + } + } + if let Some(error) = release_error { + return Err(error); + } + result +} + +async fn stage_claimed( + disks: &[EcstoreDiskStore], + candidate: &PendingMigration, + owner: Uuid, + limits: MigrationLimits, +) -> Result { + candidate.revalidate(disks, limits).await?; + candidate.validate_limits(limits)?; + let lineage = read_staging_lineage(disks, limits).await?; + let previous = lineage.latest.as_ref(); + if previous.is_some_and(|old| old.manifest.owner != owner) { + return Err(MigrationError::Conflict); + } + let mut candidate = candidate.clone(); + let key = |source: &Source| (source.disk_id, source.path, source.digest, source.bytes.is_some()); + let mut source_index = std::collections::HashMap::new(); + for (index, source) in candidate.sources.iter().chain(&candidate.inherited).enumerate() { + if source_index.insert(key(source), index).is_some() { + return Err(MigrationError::Invalid); + } + } + if let Some(old) = previous { + for source in old.candidate.sources.iter().chain(&old.candidate.inherited) { + if let Some(index) = source_index.get(&key(source)).copied() { + let existing = if index < candidate.sources.len() { + &candidate.sources[index] + } else { + &candidate.inherited[index - candidate.sources.len()] + }; + if existing != source { + return Err(MigrationError::Conflict); + } + } else { + if source_index.len() >= limits.max_sources { + return Err(SnapshotError::TooLarge.into()); + } + source_index.insert(key(source), candidate.sources.len() + candidate.inherited.len()); + candidate.inherited.push(source.clone()); + } + } + } + if candidate.replay_records(limits)?.is_empty() { + return Err(MigrationError::Empty); + } + let payload = candidate.encode(limits)?; + // Exact retry comparison must include all inherited responsibilities. + // Validating only the freshly captured sources would reject our own + // interrupted successor payload before it can be completed. + lineage.validate_orphans(Some(&payload))?; + let digest: [u8; 32] = Sha256::digest(&payload).into(); + let (sequence, slot) = previous.map_or((1, 0), |old| { + if old.manifest.payload_digest == digest { + (old.manifest.sequence, old.slot) + } else { + (old.manifest.sequence.saturating_add(1), 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 readback = read_staging_lineage(disks, limits).await?; + readback.validate_orphans(None)?; + let recovered = readback.latest.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], + limits: MigrationLimits, +) -> Result, MigrationError> { + let disks = configured_disks(disks).await?; + let lineage = read_staging_lineage(&disks, limits).await?; + lineage.validate_orphans(None)?; + let Some(staged) = lineage.latest 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, + max_sources: 64, + }; + + #[derive(Clone, Copy, PartialEq, Eq)] + pub(super) enum Boundary { + AfterPayload, + BeforeManifest, + AfterManifest, + AfterReadback, + Release, + } + + static INTERRUPTIONS: std::sync::LazyLock>> = + std::sync::LazyLock::new(Default::default); + + type SourceChange = (EcstoreDiskStore, Vec); + type SourceChangeMap = std::collections::BTreeMap; + + static SOURCE_CHANGES: std::sync::LazyLock> = 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) -> Vec { + 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"); + let previous = read_bounded(&disk, PAYLOADS[0], LIMITS.max_bytes).await.expect("old payload"); + 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!( + recover_pending_migration(&disks, LIMITS).await.is_err(), + "unknown successor responsibility must block recovery" + ); + assert_eq!(read_bounded(&disk, PAYLOADS[0], LIMITS.max_bytes).await.expect("old anchor"), previous); + } + } + + #[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) + )); + let empty = capture_legacy_migration(&[Some(disk.clone())], LIMITS) + .await + .expect("observed empty sources"); + assert!(matches!( + stage_legacy_migration(&[Some(disk.clone())], &empty, Uuid::new_v4(), 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; + if matches!(boundary, Boundary::AfterManifest | Boundary::AfterReadback) { + assert!(recovery.expect("committed restart").is_some()); + } else { + assert!( + matches!(recovery, Err(MigrationError::Conflict)), + "uncommitted candidate is not a committed recovery result" + ); + } + 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!(matches!(recover_pending_migration(&disks, LIMITS).await, Err(MigrationError::Conflict))); + 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)); + } + } + + async fn slot_bytes(disk: &EcstoreDiskStore) -> Vec>> { + let mut bytes = Vec::new(); + for path in PAYLOADS.into_iter().chain(COMMITS) { + bytes.push(read_bounded(disk, path, LIMITS.max_bytes).await.expect("slot bytes")); + } + bytes + } + + #[tokio::test] + async fn migration_damaged_newer_commit_never_overwrites_successor_responsibility() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + for name in ["a", "b"] { + source(&disk, &record(name, MrfKind::PartialWrite, None)).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("committed generation"); + } + install(&disk, COMMITS[1], b"torn higher manifest", MANIFEST_LEN) + .await + .expect("manifest fault"); + source(&disk, &record("c", MrfKind::PartialWrite, None)).await; + let before = slot_bytes(&disk).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("latest legacy source"); + assert!(stage_legacy_migration(&disks, &candidate, owner, LIMITS).await.is_err()); + assert_eq!(slot_bytes(&disk).await, before, "the only payload containing b must not be overwritten"); + } + + #[tokio::test] + async fn migration_lower_limits_never_fall_back_to_a_smaller_old_generation() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + for name in ["a", "b"] { + source(&disk, &record(name, MrfKind::PartialWrite, None)).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("committed generation"); + } + source(&disk, &record("a", MrfKind::PartialWrite, None)).await; + let before = slot_bytes(&disk).await; + let first_len = before[0].as_ref().expect("first payload").len(); + for limits in [ + MigrationLimits { + max_bytes: first_len, + ..LIMITS + }, + MigrationLimits { + max_records: 1, + ..LIMITS + }, + MigrationLimits { + max_sources: 2, + ..LIMITS + }, + ] { + assert!(matches!( + recover_pending_migration(&disks, limits).await, + Err(MigrationError::Snapshot(SnapshotError::TooLarge)) + )); + assert_eq!(slot_bytes(&disk).await, before); + } + } + + #[tokio::test] + async fn migration_source_history_count_is_bounded_before_successor_write() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + let limits = MigrationLimits { + max_sources: 3, + ..LIMITS + }; + let mut records = (0..10) + .map(|n| record(&n.to_string(), MrfKind::PartialWrite, None)) + .collect::>(); + for round in 0..3 { + records.rotate_left(1); + source(&disk, &records.concat()).await; + let candidate = capture_legacy_migration(&disks, limits) + .await + .expect("bounded current sources"); + assert_eq!(candidate.replay_records(limits).expect("same responsibilities").len(), 10); + let before = slot_bytes(&disk).await; + let result = stage_legacy_migration(&disks, &candidate, owner, limits).await; + if round < 2 { + assert_eq!(result.expect("within source count"), round + 1); + } else { + assert!(matches!(result, Err(MigrationError::Snapshot(SnapshotError::TooLarge)))); + assert_eq!(slot_bytes(&disk).await, before); + } + } + } + + #[tokio::test] + async fn migration_empty_legacy_sources_still_inherit_prior_responsibilities() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + let bytes = record("a", MrfKind::PartialWrite, None); + source(&disk, &bytes).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("initial source"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("initial stage"); + EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + MRF_SCOPED_JOURNAL_PATH, + Some(bytes.clone().into()), + None, + ) + .await + .expect("legacy removes source"); + let empty = capture_legacy_migration(&disks, LIMITS) + .await + .expect("valid absent-source observation"); + assert!(empty.replay_records(LIMITS).expect("empty observation").is_empty()); + assert_eq!( + stage_legacy_migration(&disks, &empty, owner, LIMITS) + .await + .expect("inherit earlier obligation"), + 2 + ); + assert_eq!( + recover_pending_migration(&disks, LIMITS) + .await + .expect("restart") + .expect("pending") + .replay_records(LIMITS) + .expect("inherited responsibility"), + vec![bytes] + ); + } + + #[tokio::test] + async fn migration_release_failure_still_releases_other_owned_claims() { + let root = TempDir::new().expect("test directory"); + let first = disk(&root, "first").await; + let second = disk(&root, "second").await; + source(&first, &record("a", MrfKind::PartialWrite, None)).await; + source(&second, &record("a", MrfKind::PartialWrite, None)).await; + let disks = [Some(first), Some(second)]; + let ordered = configured_disks(&disks).await.expect("disk order"); + let owner = Uuid::new_v4(); + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture"); + INTERRUPTIONS.lock().expect("fault map").insert(owner, Boundary::Release); + assert!(stage_legacy_migration(&disks, &candidate, owner, LIMITS).await.is_err()); + assert!( + read_bounded(&ordered[0], CLAIM, 64) + .await + .expect("failed release remains claimed") + .is_some() + ); + assert!( + read_bounded(&ordered[1], CLAIM, 64) + .await + .expect("later release attempted") + .is_none() + ); + } + + #[tokio::test] + async fn migration_retry_repairs_missing_manifest_replica() { + let root = TempDir::new().expect("test directory"); + let first = disk(&root, "first").await; + let second = disk(&root, "second").await; + let disks = [Some(first.clone()), Some(second.clone())]; + let owner = Uuid::new_v4(); + for name in ["a", "b"] { + let bytes = record(name, MrfKind::PartialWrite, None); + source(&first, &bytes).await; + source(&second, &bytes).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture generation"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("stage generation"); + } + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture"); + let committed = read_bounded(&second, COMMITS[1], MANIFEST_LEN) + .await + .expect("manifest") + .expect("committed"); + EcstoreDiskAPI::compare_and_update_file( + second.as_ref(), + RUSTFS_META_BUCKET, + COMMITS[1], + Some(committed.clone().into()), + None, + ) + .await + .expect("lost replica"); + assert_eq!( + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("retry missing replica"), + 2 + ); + assert_eq!( + read_bounded(&second, COMMITS[1], MANIFEST_LEN) + .await + .expect("repaired manifest"), + Some(committed) + ); + } + + #[tokio::test] + async fn migration_old_payload_orphan_matches_any_validated_replica() { + let root = TempDir::new().expect("test directory"); + let first = disk(&root, "first").await; + let second = disk(&root, "second").await; + let disks = [Some(first.clone()), Some(second)]; + let owner = Uuid::new_v4(); + for name in ["a", "b"] { + let bytes = record(name, MrfKind::PartialWrite, None); + for disk in disks.iter().flatten() { + source(disk, &bytes).await; + } + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture generation"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("stage generation"); + } + let old_manifest = read_bounded(&first, COMMITS[0], MANIFEST_LEN) + .await + .expect("old manifest") + .expect("gen1"); + assert_eq!( + EcstoreDiskAPI::compare_and_update_file( + first.as_ref(), + RUSTFS_META_BUCKET, + COMMITS[0], + Some(old_manifest.into()), + None + ) + .await + .expect("remove one old manifest"), + EcstoreConditionalFileUpdate::Updated + ); + let before = slot_bytes(&first).await; + for ordered in [disks.clone(), [disks[1].clone(), disks[0].clone()]] { + let recovered = recover_pending_migration(&ordered, LIMITS) + .await + .expect("older orphan has independent proof") + .expect("gen2"); + assert_eq!(recovered.replay_records(LIMITS).expect("A and B obligations").len(), 2); + let candidate = capture_legacy_migration(&ordered, LIMITS).await.expect("current B source"); + assert_eq!( + stage_legacy_migration(&ordered, &candidate, owner, LIMITS) + .await + .expect("gen2 retry"), + 2 + ); + assert_eq!( + slot_bytes(&first).await, + before, + "known older orphan must not cause fallback or overwrite" + ); + } + } + + #[tokio::test] + async fn migration_successor_retry_validates_orphan_after_inheriting_previous_records() { + for boundary in [Boundary::AfterPayload, Boundary::BeforeManifest] { + 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("source A"); + stage_legacy_migration(&disks, &original, owner, LIMITS).await.expect("gen1"); + let old_slot = slot_bytes(&disk).await; + source(&disk, &b).await; + let next = capture_legacy_migration(&disks, LIMITS).await.expect("source B"); + INTERRUPTIONS.lock().expect("fault map").insert(owner, boundary); + assert!(stage_legacy_migration(&disks, &next, owner, LIMITS).await.is_err()); + assert!( + matches!(recover_pending_migration(&disks, LIMITS).await, Err(MigrationError::Conflict)), + "uncommitted AB is not silently accepted as A" + ); + let captured_again = capture_legacy_migration(&disks, LIMITS) + .await + .expect("restart source capture contains only B"); + assert_eq!(captured_again.replay_records(LIMITS).expect("current source"), vec![b.clone()]); + assert_eq!( + stage_legacy_migration(&disks, &captured_again, owner, LIMITS) + .await + .expect("retry must compare inherited AB"), + 2 + ); + let recovered = recover_pending_migration(&disks, LIMITS) + .await + .expect("committed restart") + .expect("gen2"); + let records = recovered.replay_records(LIMITS).expect("retained A and B"); + assert_eq!(records.len(), 2); + assert!(records.contains(&a) && records.contains(&b)); + let after = slot_bytes(&disk).await; + assert_eq!(after[0], old_slot[0]); + assert_eq!(after[2], old_slot[2]); + } + } + + #[tokio::test] + async fn migration_third_generation_retry_keeps_reused_slot_recoverable() { + for boundary in [Boundary::AfterPayload, Boundary::BeforeManifest] { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + let records = ["a", "b", "c"] + .into_iter() + .map(|name| record(name, MrfKind::PartialWrite, None)) + .collect::>(); + + for bytes in &records[..2] { + source(&disk, bytes).await; + let candidate = capture_legacy_migration(&disks, LIMITS) + .await + .expect("capture committed generation"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("stage committed generation"); + } + + source(&disk, &records[2]).await; + let third = capture_legacy_migration(&disks, LIMITS) + .await + .expect("capture third generation"); + INTERRUPTIONS.lock().expect("fault map").insert(owner, boundary); + assert!(stage_legacy_migration(&disks, &third, owner, LIMITS).await.is_err()); + assert!(matches!(recover_pending_migration(&disks, LIMITS).await, Err(MigrationError::Conflict))); + + let retry = capture_legacy_migration(&disks, LIMITS) + .await + .expect("recapture third generation"); + assert_eq!( + stage_legacy_migration(&disks, &retry, owner, LIMITS) + .await + .expect("retry third generation"), + 3 + ); + let recovered = recover_pending_migration(&disks, LIMITS) + .await + .expect("recover after third retry") + .expect("third generation"); + let replayed = recovered.replay_records(LIMITS).expect("all staged responsibilities"); + assert_eq!(replayed.len(), 3); + for record in &records { + assert!(replayed.contains(record)); + } + } + } + + #[tokio::test] + async fn migration_third_generation_source_change_retry_keeps_reused_slot_recoverable() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + let records = ["a", "b", "c", "d"] + .into_iter() + .map(|name| record(name, MrfKind::PartialWrite, None)) + .collect::>(); + + for bytes in &records[..2] { + source(&disk, bytes).await; + let candidate = capture_legacy_migration(&disks, LIMITS) + .await + .expect("capture committed generation"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("stage committed generation"); + } + + source(&disk, &records[2]).await; + let third = capture_legacy_migration(&disks, LIMITS) + .await + .expect("capture third generation"); + SOURCE_CHANGES + .lock() + .expect("source fault map") + .insert(owner, (disk.clone(), records[3].clone())); + assert!(matches!( + stage_legacy_migration(&disks, &third, owner, LIMITS).await, + Err(MigrationError::SourceChanged) + )); + assert!(matches!(recover_pending_migration(&disks, LIMITS).await, Err(MigrationError::Conflict))); + + source(&disk, &records[2]).await; + let retry = capture_legacy_migration(&disks, LIMITS) + .await + .expect("recapture restored third generation"); + assert_eq!( + stage_legacy_migration(&disks, &retry, owner, LIMITS) + .await + .expect("retry restored third generation"), + 3 + ); + let recovered = recover_pending_migration(&disks, LIMITS) + .await + .expect("recover after restored third retry") + .expect("third generation"); + let replayed = recovered.replay_records(LIMITS).expect("all staged responsibilities"); + assert_eq!(replayed.len(), 3); + for record in &records[..3] { + assert!(replayed.contains(record)); + } + assert!(!replayed.contains(&records[3])); + } + + #[tokio::test] + async fn migration_newer_manifest_with_stale_payload_fails_closed() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let disks = [Some(disk.clone())]; + let owner = Uuid::new_v4(); + for name in ["a", "b"] { + source(&disk, &record(name, MrfKind::PartialWrite, None)).await; + let candidate = capture_legacy_migration(&disks, LIMITS).await.expect("capture"); + stage_legacy_migration(&disks, &candidate, owner, LIMITS) + .await + .expect("commit"); + } + let old = read_bounded(&disk, PAYLOADS[0], LIMITS.max_bytes) + .await + .expect("old read") + .expect("old payload"); + install(&disk, PAYLOADS[1], &old, LIMITS.max_bytes) + .await + .expect("stale payload corruption"); + source(&disk, &record("a", MrfKind::PartialWrite, None)).await; + let recovery = recover_pending_migration(&disks, LIMITS).await; + assert!( + recovery.is_err(), + "newer manifest must prevent fallback to old responsibilities: {recovery:?}" + ); + } +} diff --git a/docs/testing/mrf-pending-migration.md b/docs/testing/mrf-pending-migration.md new file mode 100644 index 000000000..f07638c1c --- /dev/null +++ b/docs/testing/mrf-pending-migration.md @@ -0,0 +1,21 @@ +# Pending MRF Migration + +`heal::mrf_queue::snapshot::migration` exposes explicit capture, staging, and readback of pending legacy responsibility evidence. Nothing invokes it from the production MRF consumer. It does not enable the committed-snapshot writer, freeze legacy ingress, acknowledge durable admission, or authorize source garbage collection. + +The caller supplies every configured local disk slot, including missing slots. Missing/unformatted/duplicate disks, unavailable metadata volumes, invalid records, and aggregate byte/record/source-history overflow fail closed. A valid empty source observation can inherit earlier pending responsibilities, but staging without any current or inherited responsibility is rejected. Both legacy paths retain their original bytes, disk identity, absent-versus-empty state, and SHA-256 digest. Complete subset/superset replicas become conservative pending evidence, never a claimed newest legacy snapshot. Raw record replay preserves kind, scope and nil/absent version encodings; unknown incarnation stays unknown. + +Staging writes only `.heal-mrf-import-pending.{0,1}.bin`, `.heal-mrf-import-commit.{0,1}.bin`, and `.heal-mrf-import-claim.bin` under the metadata volume. It reuses the committed reader's manifest codec and the storage owner's conditional-file operation, including the configured metadata durability policy. A candidate is written before sources are revalidated, its manifest is then committed, and committed bytes plus source coverage are read back before success. Success is pending staging evidence, not a power-loss or cluster-quorum durability receipt. + +A successor inherits prior source bytes even if a replay consumer has already read or admitted their records. Independent byte, record, and source-history limits include inherited evidence, and source identities use a hash index with full-byte conflict checks. There is no completion-based pruning. A changed source blocks recovery of that pending generation; an explicit new capture can stage a successor that retains both the previous and current responsibilities. The inactive slot is replaced while the preceding committed slot remains intact. Any corrupt, unsupported or over-budget slot blocks selection rather than falling back to an older generation. The reader first collects bounded lineage evidence from at most two slots per configured disk. A payload without a manifest is repairable only when its length and digest match the complete retry candidate after inheritance, or any independently validated committed payload, including an older generation on another replica. Unknown payloads still block writing. Retries also repair missing replica manifests. + +All participating disks are claimed in disk-identity order through CAS. Normal completion conditionally releases only the current invocation's claim, attempts every acquired claim even after a release error, and reports the first release error. Cancellation, process death, or an ambiguous claim/release I/O failure may leave a claim behind. Read-only recovery remains available when the snapshot/source proof is valid, but further staging is blocked until a separate storage-fenced recovery procedure is implemented. Process liveness and the legacy ingress lease do not authorize taking over or deleting a claim. + +Run the focused fixtures with a nonzero test count: + +```sh +cargo test -p rustfs-heal --lib heal::mrf_queue::snapshot::migration::tests +``` + +Fixtures use real local disks and the production CAS/readback path. They cover disk-order independence, retained raw identities, source change after candidate write, capacity rejection, interrupted commit boundaries, lost responses, torn inactive slots, and refusal to take over an interrupted claim. Boundary injection and same-process reopen are not process-kill, directory-fsync failure, disk-full, mixed-version, or power-loss tests. The actual manager Full/Accepted-to-crash pipeline remains outside this staged API. + +Rollback leaves all pending and legacy artifacts untouched. Activation still requires legacy-writer coordination, bounded recoverable ingress, exact object-disposition/successor receipts, and the W14/W21 process-crash and compatibility gates. The production legacy replay deletion window remains unresolved by this staging-only phase.