diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 3cd199a9f..6175d9583 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1567,6 +1567,18 @@ impl HealManager { .await } + pub(crate) async fn durable_mrf_repair_anchor( + &self, + intent: &rustfs_common::mrf_channel::MrfIntent, + ) -> Option { + match self.storage.mrf_bucket_incarnation_id(intent.bucket.as_ref()).await { + Ok(Some(bucket_incarnation_id)) => { + rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, bucket_incarnation_id) + } + Ok(None) | Err(_) => None, + } + } + async fn submit_heal_request_with_receipt_alias_and_mrf_notice( &self, request: HealRequest, diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index c9ea5a8a7..1419f5587 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -733,23 +733,30 @@ pub(super) fn mrf_verified_repair_event_for_target( HealObjectDisposition::AuthoritativelyAbsent => MrfVerifiedRepairDisposition::AuthoritativelyAbsent, _ => return None, }; - if target.kind != MrfKind::PartialWrite { - return None; - } - let expected_kind = HealObjectKind::Object; + let expected_kind = match target.kind { + MrfKind::DecodeFailure => HealObjectKind::Decode, + MrfKind::MetadataCorruption => HealObjectKind::Metadata, + MrfKind::PartialWrite => HealObjectKind::Object, + }; if outcome.identity.kind != expected_kind || outcome.identity.bucket.as_str() != target.bucket.as_ref() || outcome.identity.object.as_str() != target.object.as_ref() { return None; } - let version_id = target.version_id.filter(|bytes| *bytes != [0; 16]); + let version_id = (!matches!(target.kind, MrfKind::MetadataCorruption)) + .then_some(target.version_id) + .flatten() + .filter(|bytes| *bytes != [0; 16]); let expected_version = version_id.map(|bytes| uuid::Uuid::from_bytes(bytes).to_string()); if outcome.identity.version_id != expected_version { return None; } - let expected_pool = target.scope.and_then(|scope| usize::try_from(scope.pool_index).ok()); - let expected_set = target.scope.and_then(|scope| usize::try_from(scope.set_index).ok()); + let scope = (!matches!(target.kind, MrfKind::MetadataCorruption)) + .then_some(target.scope) + .flatten(); + let expected_pool = scope.and_then(|scope| usize::try_from(scope.pool_index).ok()); + let expected_set = scope.and_then(|scope| usize::try_from(scope.set_index).ok()); if outcome.identity.pool_index != expected_pool || outcome.identity.set_index != expected_set { return None; } @@ -759,7 +766,7 @@ pub(super) fn mrf_verified_repair_event_for_target( bucket: target.bucket.clone(), object: target.object.clone(), version_id, - scope: target.scope, + scope, lease: target.lease, bucket_incarnation_id, disposition, diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 37d5553af..6e551b829 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -1142,18 +1142,48 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() { assert_eq!(event.bucket_incarnation_id, incarnation); assert_eq!(event.disposition, MrfVerifiedRepairDisposition::Repaired); - assert!( - mrf_verified_repair_event_for_target( - &MrfRepairNoticeTarget { - kind: MrfKind::DecodeFailure, - ..target.clone() - }, - &matching - ) - .is_none(), - "only receipt-producing partial-write object heals can publish verified events today" + let decode_target = MrfRepairNoticeTarget { + kind: MrfKind::DecodeFailure, + ..target.clone() + }; + let decode_outcome = HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Decode, + ..matching.identity.clone() + }, + ..matching.clone() + }; + let decode_event = + mrf_verified_repair_event_for_target(&decode_target, &decode_outcome).expect("decode repairs publish exact proofs"); + assert_eq!(decode_event.kind, MrfKind::DecodeFailure); + assert_eq!( + decode_event.scope, + Some(MrfScope { + pool_index: 1, + set_index: 2 + }) ); + let metadata_target = MrfRepairNoticeTarget { + kind: MrfKind::MetadataCorruption, + ..target.clone() + }; + let metadata_outcome = HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Metadata, + version_id: None, + pool_index: None, + set_index: None, + ..matching.identity.clone() + }, + ..matching.clone() + }; + let metadata_event = + mrf_verified_repair_event_for_target(&metadata_target, &metadata_outcome).expect("metadata repairs publish exact proofs"); + assert_eq!(metadata_event.kind, MrfKind::MetadataCorruption); + assert_eq!(metadata_event.version_id, None); + assert_eq!(metadata_event.scope, None); + for rejected in [ HealObjectOutcome { disposition: HealObjectDisposition::Unknown, diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 375b639fe..48ae146ce 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -36,7 +36,7 @@ use super::{DiskStore, HealDiskExt as _, local_disk_map_read}; use crate::heal::manager::{HealManager, MrfRepairNoticeTarget}; use metrics::{counter, gauge}; -use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIngressResult, MrfIntent}; +use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfDurableRepairAnchor, MrfIngressResult, MrfIntent}; use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; use std::collections::{HashSet, VecDeque}; use std::sync::Arc; @@ -527,9 +527,12 @@ struct MrfRuntime { /// True while a journal snapshot exists on disk that may still be needed /// for replay or cleanup. journal_on_disk: bool, - /// True after replay admitted records into the manager and the startup - /// journal must stay until a durable replay proof is persisted. + /// True when replay observed a responsibility that cannot be discharged by + /// a complete verified repair proof in this process. retain_replay_journal: bool, + /// Partial-write responsibilities accepted from replay and waiting for an + /// exact storage-owned proof before the startup journal can be deleted. + durable_replay_anchors: Vec, /// Earliest instant a full-admission retry may proceed. backoff_until: Option, } @@ -627,6 +630,29 @@ impl MrfRuntime { gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(self.queue.depth())); gauge!("rustfs_heal_mrf_queue_bytes").set(metric_f64(self.queue.bytes())); } + + fn retained_replay_journal(&self) -> bool { + self.retain_replay_journal || !self.durable_replay_anchors.is_empty() + } + + fn discharge_durable_replay_anchors(&mut self) { + if self.durable_replay_anchors.is_empty() { + return; + } + let mut buckets: Vec> = self + .durable_replay_anchors + .iter() + .map(|anchor| anchor.bucket.clone()) + .collect(); + buckets.sort_unstable(); + buckets.dedup(); + for bucket in buckets { + rustfs_common::mrf_channel::consume_recorded_verified_mrf_repair_events_for( + bucket.as_ref(), + &mut self.durable_replay_anchors, + ); + } + } } /// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and @@ -677,10 +703,71 @@ struct ReplayOutcome { replayed: usize, journal_on_disk: bool, retain_journal_for_replay: bool, + durable_replay_anchors: Vec, } -fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize, accepted_or_merged: bool) -> bool { - rearm_incomplete || pending_depth > 0 || accepted_or_merged +fn replay_must_retain_journal( + rearm_incomplete: bool, + pending_depth: usize, + accepted_without_durable_anchor: bool, + durable_replay_anchors: usize, +) -> bool { + rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0 +} + +#[derive(Clone, Copy)] +enum ReplayCleanup { + Legacy, + Committed { sequence: u64 }, +} + +struct ReplaySource { + data: Vec, + cleanup: ReplayCleanup, +} + +async fn read_replay_source(max_bytes: usize) -> Result, snapshot::SnapshotError> { + if let Some(committed) = snapshot::inspect_local_committed_snapshot(max_bytes).await? { + return Ok(Some(ReplaySource { + data: committed.payload().to_vec(), + cleanup: ReplayCleanup::Committed { + sequence: committed.sequence(), + }, + })); + } + // The scoped file is a complete authoritative legacy snapshot. Fall back + // to the v1 mirror only when the authoritative path is unavailable; + // merging both files could combine records from different flush epochs. + let data = match read_journal(MRF_SCOPED_JOURNAL_PATH).await { + Some(data) => data, + None => match read_journal(MRF_JOURNAL_PATH).await { + Some(data) => data, + None => return Ok(None), + }, + }; + Ok(Some(ReplaySource { + data, + cleanup: ReplayCleanup::Legacy, + })) +} + +async fn delete_replay_source(cleanup: ReplayCleanup, max_bytes: usize) -> bool { + let committed_deleted = match cleanup { + ReplayCleanup::Legacy => true, + ReplayCleanup::Committed { sequence } => match snapshot::delete_committed_snapshots_through(sequence, max_bytes).await { + Ok(deleted) => deleted, + Err(err) => { + tracing::warn!( + target: "rustfs::heal::mrf", + error = %err, + sequence, + "MRF committed replay checkpoint cleanup failed" + ); + false + } + }, + }; + committed_deleted && delete_journals().await } /// Shared replay core: read + decode + re-arm, then drain what fits. The @@ -691,22 +778,32 @@ async fn replay_into( queue: &mut MrfQueue, backoff_until: &mut Option, ) -> ReplayOutcome { - // The scoped file is a complete authoritative snapshot. Fall back to the - // legacy mirror only when the authoritative path is unavailable; merging - // both files could combine records from different flush epochs. - let data = match read_journal(MRF_SCOPED_JOURNAL_PATH).await { - Some(data) => data, - None => match read_journal(MRF_JOURNAL_PATH).await { - Some(data) => data, - None => { - return ReplayOutcome { - replayed: 0, - journal_on_disk: false, - retain_journal_for_replay: false, - }; - } - }, + let source = match read_replay_source(queue.byte_budget).await { + Ok(Some(source)) => source, + Ok(None) => { + return ReplayOutcome { + replayed: 0, + journal_on_disk: false, + retain_journal_for_replay: false, + durable_replay_anchors: Vec::new(), + }; + } + Err(err) => { + tracing::warn!( + target: "rustfs::heal::mrf", + error = %err, + "MRF committed replay checkpoint could not be inspected" + ); + return ReplayOutcome { + replayed: 0, + journal_on_disk: true, + retain_journal_for_replay: true, + durable_replay_anchors: Vec::new(), + }; + } }; + let cleanup = source.cleanup; + let data = source.data; let (decoded, truncated) = decode_journal(&data); let replayed = decoded.len(); let intents = decoded; @@ -727,7 +824,8 @@ async fn replay_into( // prefix. queue.raise_limits_for_replay(intents.len(), replay_bytes); let mut rearm_incomplete = false; - let mut accepted_or_merged = false; + let mut accepted_without_durable_anchor = false; + let mut durable_replay_anchors = Vec::new(); for intent in intents { let result = queue.try_push_typed(intent.clone()); match result { @@ -755,7 +853,11 @@ async fn replay_into( } match submit_mrf_heal_request(manager, &intent).await { Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { - accepted_or_merged = true; + if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { + durable_replay_anchors.push(anchor); + } else { + accepted_without_durable_anchor = true; + } } Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { intent.attempts = intent.attempts.saturating_add(1); @@ -787,16 +889,23 @@ async fn replay_into( } } } - let retain_journal_for_replay = replay_must_retain_journal(rearm_incomplete, queue.depth(), accepted_or_merged); - let journal_on_disk = if retain_journal_for_replay { + let must_retain_journal = replay_must_retain_journal( + rearm_incomplete, + queue.depth(), + accepted_without_durable_anchor, + durable_replay_anchors.len(), + ); + let retain_journal_for_replay = rearm_incomplete || accepted_without_durable_anchor; + let journal_on_disk = if must_retain_journal { true } else { - !delete_journals().await + !delete_replay_source(cleanup, queue.byte_budget).await }; ReplayOutcome { replayed, journal_on_disk, retain_journal_for_replay, + durable_replay_anchors, } } @@ -811,6 +920,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive dirty: false, journal_on_disk: false, retain_replay_journal: false, + durable_replay_anchors: Vec::new(), backoff_until: None, }; @@ -819,6 +929,7 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; runtime.journal_on_disk = replay.journal_on_disk; runtime.retain_replay_journal = replay.retain_journal_for_replay; + runtime.durable_replay_anchors = replay.durable_replay_anchors; // Anything still pending (e.g. the manager was full and backoff armed) // must be re-persisted by the next flush before replay can delete the // startup anchor. @@ -862,11 +973,12 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive } } _ = flush_tick.tick() => { + runtime.discharge_durable_replay_anchors(); match tick_action( runtime.dirty, runtime.queue.depth(), runtime.journal_on_disk, - runtime.retain_replay_journal, + runtime.retained_replay_journal(), ) { TickAction::Flush => { runtime.flush().await; @@ -925,7 +1037,7 @@ fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool, retain_replay_j #[cfg(test)] mod tests { use super::*; - use rustfs_common::mrf_channel::{MrfIntent, MrfKind}; + use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent}; use std::sync::Arc as StdArc; fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent { @@ -966,21 +1078,69 @@ mod tests { #[test] fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() { assert!( - replay_must_retain_journal(true, 0, false), + replay_must_retain_journal(true, 0, false, 0), "a rejected replay record still needs its disk anchor" ); assert!( - replay_must_retain_journal(false, 1, false), + replay_must_retain_journal(false, 1, false, 0), "a Full admission retry must keep the startup journal until the next snapshot" ); assert!( - !replay_must_retain_journal(false, 0, false), + !replay_must_retain_journal(false, 0, false, 0), "a fully consumed replay snapshot without accepts may be deleted" ); assert!( - replay_must_retain_journal(false, 0, true), - "accepted or merged replay records still need a durable successor" + replay_must_retain_journal(false, 0, true, 0), + "accepted or merged replay records without a proof identity still need a durable successor" ); + assert!( + replay_must_retain_journal(false, 0, false, 1), + "accepted replay records with a durable proof anchor must retain the journal until proof arrives" + ); + } + + #[test] + fn runtime_releases_retained_replay_journal_after_verified_repair_proof() { + let mut intent = intent("proof-bucket", "proof-object", 0); + intent.kind = MrfKind::PartialWrite; + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut intent), + MrfIngressResult::Enqueued + ); + let bucket_incarnation_id = uuid::Uuid::new_v4(); + let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, bucket_incarnation_id) + .expect("fresh replay lease and bucket incarnation build a durable anchor"); + let mut runtime = MrfRuntime { + queue: MrfQueue::new(2, usize::MAX), + config: MrfConsumerConfig::default(), + new_since_flush: 0, + dirty: false, + journal_on_disk: true, + retain_replay_journal: false, + durable_replay_anchors: vec![anchor], + backoff_until: None, + }; + rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent { + kind: intent.kind, + bucket: intent.bucket.clone(), + object: intent.object.clone(), + version_id: intent.version_id, + scope: intent.scope, + lease: intent.lease, + bucket_incarnation_id, + disposition: MrfVerifiedRepairDisposition::Repaired, + }); + + assert!( + runtime.retained_replay_journal(), + "anchor must retain the startup journal before proof is consumed" + ); + runtime.discharge_durable_replay_anchors(); + assert!( + !runtime.retained_replay_journal(), + "matching verified proof discharges the durable replay anchor" + ); + rustfs_common::mrf_channel::release_mrf_intent(&intent); } #[test] diff --git a/crates/heal/src/heal/mrf_queue/snapshot.rs b/crates/heal/src/heal/mrf_queue/snapshot.rs index 07aca8b09..43fc7372b 100644 --- a/crates/heal/src/heal/mrf_queue/snapshot.rs +++ b/crates/heal/src/heal/mrf_queue/snapshot.rs @@ -17,11 +17,13 @@ //! Each of two slots has a payload and a commit manifest. The manifest binds //! the writer identity, persistent sequence, length and whole-payload digest. //! Replacing the inactive slot must leave the previous committed slot intact. -//! Production publication and reclamation are deliberately not enabled here. +//! Production publication is deliberately reader-first; reclamation only +//! removes manifest entries after the owning replay path has discharged every +//! responsibility through a newer durable snapshot or a verified repair proof. //! An unreadable commit path cannot prove that only legacy data exists. This //! explicit inspection API fails closed and never mutates recovery anchors. -//! It is not wired into the legacy consumer: that transition requires the -//! ownership-aware replay and producer handoff before writer activation. +//! It is wired into the replay reader before writer activation, but the writer +//! remains gated on ownership-aware handoff. //! One surviving committed replica supports process restart recovery only; //! this reader does not establish a replication quorum or a power-loss policy. @@ -542,6 +544,91 @@ pub async fn inspect_local_recovery_snapshot(max_bytes: usize) -> Result Result, SnapshotError> { + read_committed(&super::journal_disks().await, max_bytes).await +} + +/// Remove committed manifests whose sequence is no newer than +/// `committed_through`. +/// +/// Payload files are intentionally left as orphans after their manifest is +/// removed. Readers cannot discover a payload without its matching manifest, +/// and deleting manifests first prevents an older retained slot from becoming +/// visible again after the newest replay has been fully discharged. +pub async fn delete_committed_snapshots_through(committed_through: u64, max_bytes: usize) -> Result { + let disks = super::journal_disks().await; + delete_committed_snapshots_through_on(&disks, committed_through, max_bytes).await +} + +async fn delete_committed_snapshots_through_on( + disks: &[EcstoreDiskStore], + committed_through: u64, + max_bytes: usize, +) -> Result { + if disks.is_empty() { + return Err(SnapshotError::NoWritableReplica); + } + let mut any_changed = false; + let mut first_error = None; + for disk in disks { + for path in MANIFEST_PATHS { + let existing = match read_bounded(disk, path, MANIFEST_LEN).await { + Ok(Some(existing)) => existing, + Ok(None) => continue, + Err(error) => { + if first_error.is_none() { + first_error = Some(error); + } + continue; + } + }; + let manifest = match Manifest::decode(&existing, max_bytes) { + Ok(manifest) => manifest, + Err(error) => { + if first_error.is_none() { + first_error = Some(error); + } + continue; + } + }; + if manifest.sequence > committed_through { + continue; + } + match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + path, + Some(EcstoreDiskBytes::from(existing)), + None, + ) + .await + { + Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true, + Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {} + Err(error) => { + if first_error.is_none() { + first_error = Some(SnapshotError::Disk(error)); + } + } + } + } + } + if any_changed { + Ok(true) + } else if let Some(error) = first_error { + Err(error) + } else { + Ok(true) + } +} + async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result, SnapshotError> { if let Some(snapshot) = read_committed(disks, limit).await? { return Ok(Some(RecoverySnapshot::Committed(snapshot))); @@ -1185,6 +1272,44 @@ mod tests { ); } + #[tokio::test] + async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() { + let root = TempDir::new().expect("test directory"); + let disk = disk(&root, "disk").await; + let owner = Uuid::new_v4(); + let older = payload("older"); + let newer = payload("newer"); + commit(&disk, 0, owner, 3, &older).await; + commit(&disk, 1, owner, 4, &newer).await; + + assert!( + delete_committed_snapshots_through_on(std::slice::from_ref(&disk), 3, 4096) + .await + .expect("delete old manifest"), + "old committed manifest should be removed" + ); + assert!( + matches!( + EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0]).await, + Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound) + ), + "old manifest is gone, so the old payload cannot become visible again" + ); + assert_eq!( + EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]) + .await + .expect("old payload orphan may remain") + .as_ref(), + older + ); + let recovered = read_committed(std::slice::from_ref(&disk), 4096) + .await + .expect("read newer commit") + .expect("newer commit remains visible"); + assert_eq!(recovered.sequence(), 4); + assert_eq!(recovered.payload(), newer); + } + #[tokio::test] async fn legacy_import_requires_complete_consistent_replicas() { let root = TempDir::new().expect("test directory"); diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 962ee84b9..1dc6fdddc 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -356,6 +356,14 @@ pub trait HealStorageAPI: Send + Sync { /// Get bucket info async fn get_bucket_info(&self, bucket: &str) -> Result>; + /// Return the current bucket incarnation for exact MRF durable proof + /// matching. Alternate backends that cannot expose this must return + /// `None`, leaving replay anchors retained instead of acknowledged with an + /// incomplete identity. + async fn mrf_bucket_incarnation_id(&self, _bucket: &str) -> Result> { + Ok(None) + } + /// Aggregate usage-cache baselines for the requested buckets. async fn erasure_set_usage_baseline(&self, _buckets: &[String]) -> Result> { Ok(None) @@ -815,6 +823,14 @@ impl HealStorageAPI for ECStoreHealStorage { } } + async fn mrf_bucket_incarnation_id(&self, bucket: &str) -> Result> { + self.ecstore + .bucket_incarnation_id(bucket) + .await + .map(Some) + .map_err(Error::Storage) + } + async fn erasure_set_usage_baseline(&self, buckets: &[String]) -> Result> { if buckets.is_empty() { return Ok(None); diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index 8323995ef..8bf51f72c 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -29,6 +29,7 @@ use rustfs_heal::heal::{ storage::{ECStoreHealStorage, HealStorageAPI}, }; use serial_test::serial; +use sha2::{Digest, Sha256}; #[cfg(unix)] use std::{ fs::{File, OpenOptions}, @@ -48,6 +49,10 @@ use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, Pool const META_BUCKET: &str = ".rustfs.sys"; const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin"; +const COMMITTED_PAYLOAD_REL: &str = ".heal-mrf-snapshot.0.bin"; +const COMMITTED_MANIFEST_REL: &str = ".heal-mrf-commit.0.bin"; +const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01"; +const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32; async fn heal_env() -> (Vec, Arc) { heal_env_at(None).await @@ -195,6 +200,29 @@ fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { write_journal_path_to_disks(disk_paths, JOURNAL_REL, data); } +fn committed_manifest(owner: uuid::Uuid, sequence: u64, payload: &[u8]) -> Vec { + let mut manifest = Vec::with_capacity(COMMITTED_MANIFEST_LEN); + manifest.extend_from_slice(COMMITTED_MAGIC); + manifest.push(1); + manifest.extend_from_slice(owner.as_bytes()); + manifest.extend_from_slice(&sequence.to_le_bytes()); + manifest.extend_from_slice( + &u64::try_from(payload.len()) + .expect("fixture payload length fits") + .to_le_bytes(), + ); + manifest.extend_from_slice(&Sha256::digest(payload)); + manifest.extend_from_slice(&Sha256::digest(&manifest)); + assert_eq!(manifest.len(), COMMITTED_MANIFEST_LEN, "committed fixture manifest length"); + manifest +} + +fn write_committed_snapshot_to_disks(disk_paths: &[std::path::PathBuf], sequence: u64, payload: &[u8]) { + let manifest = committed_manifest(uuid::Uuid::new_v4(), sequence, payload); + write_journal_path_to_disks(disk_paths, COMMITTED_PAYLOAD_REL, payload); + write_journal_path_to_disks(disk_paths, COMMITTED_MANIFEST_REL, &manifest); +} + fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: &str) -> bool { disk_paths .iter() @@ -255,11 +283,12 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() { } /// A journal left behind by a previous process must be replayed into the -/// manager queue and then removed, and a torn tail must not block replay of -/// the intact records. +/// manager queue, and a torn tail must not block replay of the intact records. +/// The partial-write record keeps the legacy journal as the durable anchor +/// until an exact verified repair proof can discharge it. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[serial] -async fn journal_replay_arms_intents_and_deletes_the_file() { +async fn journal_replay_arms_intents_and_retains_unproven_partial_write_anchor() { let (disk_paths, storage) = heal_env().await; // The journal reader resolves disks through the process-local disk map; @@ -285,14 +314,14 @@ async fn journal_replay_arms_intents_and_deletes_the_file() { assert!( disk_paths .iter() - .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()), - "the journal file must be removed after a successful replay" + .all(|path| Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()), + "partial-write replay must retain the legacy journal until durable proof" ); assert!( disk_paths .iter() .all(|path| !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()), - "the authoritative journal file must also be removed after replay" + "missing authoritative journal remains absent" ); let snapshot = manager.operations_snapshot().await; @@ -300,6 +329,41 @@ async fn journal_replay_arms_intents_and_deletes_the_file() { assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal"); } +/// A committed checkpoint published by the new two-slot writer is the +/// authoritative startup snapshot. Legacy mirrors are fallback-only and must +/// not be merged with or preferred over the committed epoch. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn committed_snapshot_replay_takes_precedence_over_stale_legacy_mirror() { + let (disk_paths, storage) = heal_env().await; + register_local_disks(&disk_paths, "mrf-committed-replay-test").await; + + let committed = scoped_journal_record(3, "committed-bucket", "committed-object", Some([9u8; 16]), 0, 0, 0); + let stale_legacy = journal_record(1, "legacy-bucket", "legacy-object", None, 0); + write_committed_snapshot_to_disks(&disk_paths, 7, &committed); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &stale_legacy); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &stale_legacy); + + let manager = make_manager(storage); + let replayed = mrf_queue::replay_journal_once(&manager).await; + assert_eq!(replayed, 1, "only the committed snapshot epoch may replay"); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_source.mrf, 1); + assert_eq!( + snapshot.queued_by_priority.normal, 1, + "the committed partial-write record must replay instead of the stale legacy decode-failure" + ); + assert_eq!( + snapshot.queued_by_priority.urgent, 0, + "stale legacy decode-failure records must not be mixed into committed replay" + ); + assert!( + journal_exists_on_all_disks(&disk_paths, COMMITTED_MANIFEST_REL), + "the committed checkpoint remains until the accepted partial-write has proof" + ); +} + /// A canonical snapshot and its compatibility mirror may differ after a /// partial flush. Replay must choose the complete canonical epoch instead of /// combining records that never coexisted in memory. @@ -322,21 +386,24 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() { assert_eq!(snapshot.queued_by_source.mrf, 1); assert!( disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() }), - "replay cleanup must remove both journal paths" + "accepted replay responsibilities remain anchored until a verified repair proof" ); // A scoped-only snapshot is valid during a rollout where no legacy // compatibility mirror was written. Missing legacy files must not leave // the runtime in a permanent cleanup-retry state. + let (disk_paths, storage) = heal_env().await; + register_local_disks(&disk_paths, "mrf-scoped-authoritative-test").await; + let manager = make_manager(storage); let scoped_only = journal_record(1, "scoped-only-bucket", "scoped-only-object", None, 0); write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &scoped_only); assert_eq!(mrf_queue::replay_journal_once(&manager).await, 1); assert!(disk_paths.iter().all(|path| { !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() })); let scoped_v2 = scoped_journal_record(1, "scoped-v2-bucket", "scoped-v2-object", None, 0, 3, 7); @@ -350,12 +417,12 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() { ); assert_eq!( manager.operations_snapshot().await.queued_by_source.mrf, - 3, - "only the three authoritative/scoped-only epochs should have reached the manager" + 2, + "only the scoped-only and scoped-v2 authoritative epochs should have reached the manager" ); assert!(disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() })); } @@ -400,10 +467,13 @@ async fn authoritative_journal_replay_preserves_kind_and_scope_identity() { snapshot.queued_by_priority.urgent, 1, "decode-failure repair must not merge with object repair responsibility" ); - assert!(disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() - })); + assert!( + disk_paths.iter().all(|path| { + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + }), + "partial-write responsibilities keep both replay anchors until proof" + ); } /// If replay reaches a full heal-manager queue, the old journal remains the @@ -698,10 +768,10 @@ async fn journal_replay_survives_successor_flush_before_delete() { ); assert!( disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() }), - "a fully consumed successor snapshot may be deleted after restart replay" + "the accepted successor remains anchored until a verified repair proof" ); } @@ -751,10 +821,10 @@ async fn journal_replay_survives_service_kill_after_successor_flush() { ); assert!( disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() }), - "a fully consumed successor snapshot may be deleted after service-kill restart replay" + "the accepted successor remains anchored until a verified repair proof after service-kill restart" ); } @@ -814,9 +884,9 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef ); assert!( disk_paths.iter().all(|path| { - !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() - && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() }), - "a fully consumed authoritative successor may clean both epochs after restart replay" + "the accepted authoritative successor remains anchored until a verified repair proof" ); }