diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs index 37c17ae90..69e16987f 100644 --- a/crates/common/src/mrf_channel.rs +++ b/crates/common/src/mrf_channel.rs @@ -22,8 +22,9 @@ //! hint is always preferred over stalling an IO path. Durable replay of //! unconsumed intents is the consumer's job (see `rustfs-heal` //! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`. Committed partial -//! writes use a separate bounded channel and await a checkpoint receipt from -//! [`persist_partial_write_intent`]; ordinary complete writes do not use it. +//! writes and committed partial delete-marker purges use a separate bounded +//! channel and await a checkpoint receipt; ordinary complete writes do not use +//! it. use std::collections::HashMap; use std::collections::hash_map::RandomState; @@ -46,10 +47,10 @@ const MRF_COALESCER_MAX_KEYS: usize = 8192; const MRF_COALESCER_MAX_BYTES: usize = 16 * 1024 * 1024; const MRF_COALESCER_TTL: Duration = Duration::from_secs(60); const MRF_MAX_IDENTITY_COMPONENT: usize = 1024; +pub const MRF_MAX_DELETE_MARKER_BYTES: usize = 1024 * 1024; /// Why an intent was produced. Drives the heal priority mapping on the -/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High, -/// PartialWrite -> Normal). +/// consumer side. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub enum MrfKind { /// Erasure decode failed while serving a read (read path). @@ -58,6 +59,8 @@ pub enum MrfKind { MetadataCorruption, /// A write left the object with fewer committed shards than the set size. PartialWrite, + /// A committed explicit delete-marker purge missed at least one member. + DeleteMarkerPurge, } impl MrfKind { @@ -66,8 +69,60 @@ impl MrfKind { MrfKind::DecodeFailure => "decode-failure", MrfKind::MetadataCorruption => "metadata-corruption", MrfKind::PartialWrite => "partial-write", + MrfKind::DeleteMarkerPurge => "delete-marker-purge", } } + + pub const fn is_durable(self) -> bool { + matches!(self, Self::PartialWrite | Self::DeleteMarkerPurge) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct MrfDeleteMarkerPurgeIdentity { + pub bucket_incarnation_id: Uuid, + pub marker_incarnation_id: Uuid, + pub marker_identity: [u8; 32], +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct MrfDeleteMarkerPurge { + pub bucket_incarnation_id: Uuid, + pub marker_incarnation_id: Uuid, + pub marker_identity: [u8; 32], + pub marker: Arc<[u8]>, +} + +impl MrfDeleteMarkerPurge { + pub fn new( + bucket_incarnation_id: Uuid, + marker_incarnation_id: Uuid, + marker_identity: [u8; 32], + marker: Vec, + ) -> Option { + let payload = Self { + bucket_incarnation_id, + marker_incarnation_id, + marker_identity, + marker: Arc::from(marker), + }; + payload.is_valid().then_some(payload) + } + + pub const fn identity(&self) -> MrfDeleteMarkerPurgeIdentity { + MrfDeleteMarkerPurgeIdentity { + bucket_incarnation_id: self.bucket_incarnation_id, + marker_incarnation_id: self.marker_incarnation_id, + marker_identity: self.marker_identity, + } + } + + pub fn is_valid(&self) -> bool { + !self.bucket_incarnation_id.is_nil() + && self.marker_incarnation_id == self.bucket_incarnation_id + && !self.marker.is_empty() + && self.marker.len() <= MRF_MAX_DELETE_MARKER_BYTES + } } /// One repair intent. Kept deliberately small so the in-memory queue and the @@ -80,6 +135,10 @@ pub struct MrfIntent { /// Version the intent targets, as raw UUID bytes. pub version_id: Option<[u8; 16]>, pub kind: MrfKind, + /// Present only for [`MrfKind::DeleteMarkerPurge`]. The encoded marker is a + /// destructive-operation precondition, while `marker_identity` is its + /// deterministic queue and proof identity. + pub delete_marker_purge: Option, /// Stable erasure-set scope when the producer has it. Kept optional so /// metadata corruption and legacy producers do not invent a scope. pub scope: Option, @@ -99,6 +158,7 @@ pub struct MrfDurableRepairAnchor { pub object: Arc, pub version_id: Option<[u8; 16]>, pub scope: Option, + pub delete_marker_purge: Option, pub lease: MrfIngressLease, pub bucket_incarnation_id: Uuid, } @@ -108,17 +168,22 @@ impl MrfDurableRepairAnchor { /// incarnation and the original ingress lease. Legacy replay records lack /// both pieces and therefore remain fail-closed. pub fn from_intent(intent: &MrfIntent, bucket_incarnation_id: Uuid) -> Option { + let lease = intent.lease?; + let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope); + let delete_marker_purge = canonical_delete_marker_purge_identity(intent)?; + let bucket_incarnation_id = delete_marker_purge + .map(|identity| identity.bucket_incarnation_id) + .unwrap_or(bucket_incarnation_id); if bucket_incarnation_id.is_nil() { return None; } - let lease = intent.lease?; - let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope); Some(Self { kind: intent.kind, bucket: intent.bucket.clone(), object: intent.object.clone(), version_id, scope, + delete_marker_purge, lease, bucket_incarnation_id, }) @@ -133,6 +198,7 @@ impl MrfDurableRepairAnchor { && self.object == event.object && self.version_id == event.version_id && self.scope == event.scope + && self.delete_marker_purge == event.delete_marker_purge && self.lease == lease && self.bucket_incarnation_id == event.bucket_incarnation_id } @@ -203,6 +269,7 @@ pub enum MrfDropReason { Uninitialized, Full, OversizedIdentity, + InvalidPayload, CoalescerFull, } @@ -221,7 +288,14 @@ impl MrfIntent { pub fn estimated_bytes(&self) -> usize { // Struct + strings + version bytes; buckets and objects are usually // far below this bound, so rounding up keeps the budget conservative. - 64 + self.bucket.len() + self.object.len() + 64usize + .saturating_add(self.bucket.len()) + .saturating_add(self.object.len()) + .saturating_add( + self.delete_marker_purge + .as_ref() + .map_or(0, |purge| std::mem::size_of::().saturating_add(purge.marker.len())), + ) } } @@ -245,8 +319,9 @@ pub enum MrfDurableAdmissionError { Timeout, } -/// A committed partial write awaiting a durable checkpoint, separate from the -/// best-effort read/scanner channel. Dropping the reply does not cancel repair. +/// A committed storage responsibility awaiting a durable checkpoint, separate +/// from the best-effort read/scanner channel. Dropping the reply does not +/// cancel repair. pub struct MrfDurableSubmission { pub intent: MrfIntent, pub response: oneshot::Sender>, @@ -260,36 +335,70 @@ pub fn init_durable_mrf_channel() -> Result Ok(receiver) } -/// Wait for checkpoint publication only on a committed partial-write path. -/// Every write gets a fresh lease: a previous repair of the same unversioned -/// key cannot discharge a later overwrite. An error is not durable admission. +/// Wait for checkpoint publication on a committed partial-write path. Every +/// write gets a fresh lease: a previous repair of the same unversioned key +/// cannot discharge a later overwrite. An error is not durable admission. pub async fn persist_partial_write_intent( bucket: &str, object: &str, version_id: Option, scope: MrfScope, ) -> Result<(), MrfDurableAdmissionError> { - if !mrf_delivery_enabled() { - return Err(MrfDurableAdmissionError::Disabled); - } - if bucket.is_empty() - || object.is_empty() - || bucket.len() > MRF_MAX_IDENTITY_COMPONENT - || object.len() > MRF_MAX_IDENTITY_COMPONENT - { - return Err(MrfDurableAdmissionError::InvalidIdentity); - } - let sender = GLOBAL_DURABLE_MRF_SENDER.get().ok_or(MrfDurableAdmissionError::Unavailable)?; - let mut intent = MrfIntent { + let intent = MrfIntent { bucket: Arc::from(bucket), object: Arc::from(object), version_id: canonical_version(version_id), kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: Some(scope), lease: None, enqueued_at_ms: unix_now_ms(), attempts: 0, }; + persist_durable_intent(intent).await +} + +pub async fn persist_delete_marker_purge_intent( + bucket: &str, + object: &str, + version_id: Uuid, + scope: MrfScope, + delete_marker_purge: MrfDeleteMarkerPurge, +) -> Result<(), MrfDurableAdmissionError> { + if version_id.is_nil() { + return Err(MrfDurableAdmissionError::InvalidIdentity); + } + let intent = MrfIntent { + bucket: Arc::from(bucket), + object: Arc::from(object), + version_id: canonical_version(Some(version_id)), + kind: MrfKind::DeleteMarkerPurge, + delete_marker_purge: Some(delete_marker_purge), + scope: Some(scope), + lease: None, + enqueued_at_ms: unix_now_ms(), + attempts: 0, + }; + persist_durable_intent_unconditionally(intent).await +} + +async fn persist_durable_intent(intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> { + if !mrf_delivery_enabled() { + return Err(MrfDurableAdmissionError::Disabled); + } + persist_durable_intent_unconditionally(intent).await +} + +async fn persist_durable_intent_unconditionally(mut intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> { + if intent.bucket.is_empty() + || intent.object.is_empty() + || intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT + || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT + || !valid_intent_payload(&intent) + { + return Err(MrfDurableAdmissionError::InvalidIdentity); + } + let sender = GLOBAL_DURABLE_MRF_SENDER.get().ok_or(MrfDurableAdmissionError::Unavailable)?; if try_rearm_mrf_replay_intent(&mut intent) != MrfIngressResult::Enqueued { return Err(MrfDurableAdmissionError::InvalidIdentity); } @@ -359,10 +468,27 @@ fn canonical_identity( let version_id = version_id.filter(|bytes| *bytes != [0; 16]); match kind { MrfKind::MetadataCorruption => (None, None), - MrfKind::DecodeFailure | MrfKind::PartialWrite => (version_id, scope), + MrfKind::DecodeFailure | MrfKind::PartialWrite | MrfKind::DeleteMarkerPurge => (version_id, scope), } } +fn canonical_delete_marker_purge_identity(intent: &MrfIntent) -> Option> { + match intent.kind { + MrfKind::DeleteMarkerPurge => intent + .delete_marker_purge + .as_ref() + .filter(|payload| payload.is_valid()) + .map(|payload| Some(payload.identity())), + _ if intent.delete_marker_purge.is_none() => Some(None), + _ => None, + } +} + +fn valid_intent_payload(intent: &MrfIntent) -> bool { + canonical_delete_marker_purge_identity(intent).is_some() + && (intent.kind != MrfKind::DeleteMarkerPurge || (intent.version_id.is_some() && intent.scope.is_some())) +} + fn identity_estimated_bytes(key: &MrfIdentityKey) -> usize { 64usize .saturating_add(key.bucket.len()) @@ -512,6 +638,9 @@ pub fn try_send_mrf_intent_typed( if !mrf_delivery_enabled() { return MrfIngressResult::Dropped(MrfDropReason::Disabled); } + if kind == MrfKind::DeleteMarkerPurge { + return MrfIngressResult::Dropped(MrfDropReason::InvalidPayload); + } let Some(sender) = GLOBAL_MRF_SENDER.get() else { return MrfIngressResult::Dropped(MrfDropReason::Uninitialized); }; @@ -535,6 +664,7 @@ pub fn try_send_mrf_intent_typed( object: key.object.clone(), version_id: key.version_id, kind, + delete_marker_purge: None, scope, lease: Some(lease), enqueued_at_ms: unix_now_ms(), @@ -562,15 +692,18 @@ pub fn try_send_mrf_intent_typed( /// producer coalescer key: the replay queue first deduplicates legacy records, /// then manager admission owns task-level deduplication with live producers. pub fn try_rearm_mrf_replay_intent(intent: &mut MrfIntent) -> MrfIngressResult { - if intent.lease.is_some() { - return MrfIngressResult::Enqueued; - } if intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT { return MrfIngressResult::Dropped(MrfDropReason::OversizedIdentity); } let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope); intent.version_id = version_id; intent.scope = scope; + if !valid_intent_payload(intent) { + return MrfIngressResult::Dropped(MrfDropReason::InvalidPayload); + } + if intent.lease.is_some() { + return MrfIngressResult::Enqueued; + } intent.lease = Some(MrfIngressLease::new(NEXT_MRF_LEASE.fetch_add(1, Ordering::Relaxed))); MrfIngressResult::Enqueued } @@ -635,6 +768,7 @@ pub struct MrfVerifiedRepairEvent { pub object: Arc, pub version_id: Option<[u8; 16]>, pub scope: Option, + pub delete_marker_purge: Option, pub lease: Option, pub bucket_incarnation_id: Uuid, pub disposition: MrfVerifiedRepairDisposition, @@ -736,6 +870,7 @@ mod tests { object: Arc::from("object"), version_id: Some([0u8; 16]), kind: MrfKind::DecodeFailure, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 0, @@ -800,6 +935,7 @@ mod tests { object: Arc::from("object"), version_id: Some([0; 16]), kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: Some(MrfScope { pool_index: 1, set_index: 2, @@ -839,6 +975,7 @@ mod tests { object: Arc::from("object"), version_id: Some([0; 16]), kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: Some(MrfScope { pool_index: 2, set_index: 3, @@ -873,6 +1010,7 @@ mod tests { pool_index: 4, set_index: 5, }), + delete_marker_purge: None, lease, bucket_incarnation_id: incarnation, }; @@ -882,6 +1020,7 @@ mod tests { object, version_id: anchor.version_id, scope: anchor.scope, + delete_marker_purge: None, lease: Some(lease), bucket_incarnation_id: incarnation, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -949,6 +1088,7 @@ mod tests { pool_index: 1, set_index: 2, }), + delete_marker_purge: None, lease, bucket_incarnation_id: incarnation, }; @@ -963,6 +1103,7 @@ mod tests { object: retained_anchor.object.clone(), version_id: retained_anchor.version_id, scope: retained_anchor.scope, + delete_marker_purge: None, lease: Some(retained_anchor.lease), bucket_incarnation_id: retained_anchor.bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -1076,6 +1217,7 @@ mod tests { pool_index: 2, set_index: 3, }), + delete_marker_purge: None, lease: Some(MrfIngressLease::new(42)), bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -1120,4 +1262,42 @@ async fn partial_write_durable_admission_waits_for_persistence_and_renews_lease( .send(Ok(())) .expect("second producer should await receipt"); assert_eq!(second.await.expect("second producer should complete"), Ok(())); + + let bucket_incarnation_id = Uuid::new_v4(); + let marker_identity = [7; 32]; + let purge = MrfDeleteMarkerPurge::new(bucket_incarnation_id, bucket_incarnation_id, marker_identity, vec![1, 2, 3]) + .expect("valid purge payload"); + let version_id = Uuid::new_v4(); + let purge_submit = + tokio::spawn(async move { persist_delete_marker_purge_intent("bucket", "marker", version_id, scope, purge).await }); + let submission = receiver.recv().await.expect("purge submission should arrive"); + assert_eq!(submission.intent.kind, MrfKind::DeleteMarkerPurge); + assert_eq!(submission.intent.version_id, Some(*version_id.as_bytes())); + assert_eq!(submission.intent.scope, Some(scope)); + assert_eq!( + submission + .intent + .delete_marker_purge + .as_ref() + .map(MrfDeleteMarkerPurge::identity), + Some(MrfDeleteMarkerPurgeIdentity { + bucket_incarnation_id, + marker_incarnation_id: bucket_incarnation_id, + marker_identity, + }) + ); + assert!(submission.intent.lease.is_some()); + submission.response.send(Ok(())).expect("purge producer should await receipt"); + assert_eq!(purge_submit.await.expect("purge producer should complete"), Ok(())); + + let invalid = MrfDeleteMarkerPurge { + bucket_incarnation_id, + marker_incarnation_id: Uuid::new_v4(), + marker_identity, + marker: Arc::from([1u8]), + }; + assert_eq!( + persist_delete_marker_purge_intent("bucket", "marker", version_id, scope, invalid).await, + Err(MrfDurableAdmissionError::InvalidIdentity) + ); } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index b262d3c9d..69b0a126d 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4208,6 +4208,59 @@ impl DiskHealthEntry { } impl SetDisks { + pub(in crate::set_disk) async fn persist_delete_marker_purge( + &self, + bucket: &str, + object: &str, + version: Uuid, + purge: rustfs_common::mrf_channel::MrfDeleteMarkerPurge, + ) -> bool { + use rustfs_common::mrf_channel::{MrfDurableAdmissionError, MrfScope, persist_delete_marker_purge_intent}; + let scope = (|| { + Ok::<_, MrfDurableAdmissionError>(MrfScope { + pool_index: u32::try_from(self.pool_index).map_err(|_| MrfDurableAdmissionError::InvalidIdentity)?, + set_index: u32::try_from(self.set_index).map_err(|_| MrfDurableAdmissionError::InvalidIdentity)?, + }) + })(); + let result = match scope { + Ok(scope) => persist_delete_marker_purge_intent(bucket, object, version, scope, purge).await, + Err(error) => Err(error), + }; + match result { + Ok(()) => { + tracing::trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + state = "delete_marker_purge_durably_admitted", + bucket, + object, + version_id = %version, + pool_index = self.pool_index, + set_index = self.set_index, + "Delete-marker purge responsibility persisted" + ); + true + } + Err(error) => { + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + state = "delete_marker_purge_admission_failed", + bucket, + object, + version_id = %version, + pool_index = self.pool_index, + set_index = self.set_index, + error = %error, + "Delete-marker purge responsibility could not be persisted" + ); + false + } + } + } + pub(in crate::set_disk) async fn persist_partial_write(&self, bucket: &str, object: &str, version_id: Option<&str>) -> bool { use rustfs_common::mrf_channel::{ MrfDurableAdmissionError, MrfScope, mrf_delivery_enabled, persist_partial_write_intent, @@ -15155,8 +15208,8 @@ mod tests { "a stale null-marker lifecycle target must not delete its replacement: {stale_errors:?}" ); - let (current_marker, _, current_error) = set_disks - .get_object_info_and_quorum( + let (current_marker, _, _, current_error) = set_disks + .get_object_info_fileinfo_and_quorum( bucket, object, &ObjectOptions { diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 4a1e81158..929650bbc 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -521,6 +521,77 @@ fn warn_heal_writer_failures( } impl SetDisks { + pub(crate) async fn purge_delete_marker_exact( + &self, + bucket: &str, + object: &str, + version: Uuid, + marker: rustfs_filemeta::MetaDeleteMarker, + ) -> Result { + let disks = self.get_disks_internal().await; + if disks.len() != self.set_drive_count || disks.iter().any(Option::is_none) { + return Err(StorageError::SlowDown); + } + let _object_guard = self.acquire_write_lock_diag("delete_marker_purge", bucket, object).await?; + let scope = crate::store::bucket_heal_scope(bucket).ok_or(StorageError::PreconditionFailed)?; + scope.check()?; + + let request = FileInfo { + volume: bucket.to_owned(), + name: object.to_owned(), + version_id: Some(version), + ..Default::default() + }; + let results = join_all(disks.iter().enumerate().map(|(disk_index, disk)| { + let request = request.clone(); + let marker = marker.clone(); + async move { + if let Some(error) = injected_dangling_delete_error(bucket, object, disk_index) { + return Err(error); + } + let Some(disk) = disk else { return Err(DiskError::DiskNotFound) }; + disk.delete_version( + bucket, + object, + request, + false, + DeleteOptions { + expected_delete_marker: Some(marker), + ..Default::default() + }, + ) + .await + } + })) + .await; + let mut removed = false; + for result in results { + match result { + Ok(()) => removed = true, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {} + Err(error) => return Err(error.into()), + } + } + + scope.check()?; + let (_, errors) = + Self::read_all_fileinfo(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?; + let selected = self.get_disks_internal().await; + let same_targets = selected.len() == disks.len() && selected.iter().zip(&disks).all(|(current, original)| { + matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original)) + }); + if !same_targets + || !errors + .iter() + .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound))) + { + return Err(StorageError::SlowDown); + } + scope.check()?; + self.invalidate_get_object_metadata_cache(bucket, object).await; + Ok(removed) + } + /// Read back one healed version from every explicitly admitted replacement /// target. This is intentionally separate from the normal heal result: a /// successful result describes the transaction attempt, while automatic diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index d4dacd1c8..777e19265 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -20,6 +20,7 @@ //! SetDisks core (io_primitives) via inherent calls. use crate::core::pools::DecommissionCapacityAdmission; +use rustfs_common::mrf_channel::MrfDeleteMarkerPurge; use rustfs_filemeta::metadata_keys; #[cfg(test)] @@ -63,6 +64,27 @@ use super::super::{ take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn, }; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; + +fn delete_marker_purge_candidate( + file_info: &FileInfo, + expected_bucket_incarnation_id: Option, +) -> Option { + let bucket_incarnation_id = expected_bucket_incarnation_id.filter(|id| !id.is_nil())?; + let version_id = file_info.version_id.filter(|id| !id.is_nil())?; + if !file_info.is_canonical_delete_marker() { + return None; + } + let marker_incarnation_id = file_info.delete_marker_incarnation()?; + if marker_incarnation_id != bucket_incarnation_id { + return None; + } + let marker = rustfs_filemeta::MetaDeleteMarker::from(file_info.clone()); + if marker.version_id != Some(version_id) { + return None; + } + let marker_identity = marker.stable_identity(); + MrfDeleteMarkerPurge::new(bucket_incarnation_id, marker_incarnation_id, marker_identity, marker.marshal_msg().ok()?) +} use crate::api::config::storageclass; use crate::bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps; use crate::bucket::utils::is_meta_bucketname; @@ -7529,6 +7551,168 @@ impl SetDisks { Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } + #[tracing::instrument(skip(self, delete_marker_purge))] + async fn delete_object_version_with_purge( + &self, + bucket: &str, + object: &str, + fi: &FileInfo, + force_del_marker: bool, + delete_marker_purge: Option, + ) -> Result<()> { + let transported = delete_file_info_with_replication_transport_metadata(fi); + let fi = &transported; + let disks = self.disk_inventory().await; + let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit()); + let write_quorum = disks.len() / 2 + 1; + let rollback_dir = Uuid::new_v4(); + + let mut futures = Vec::with_capacity(disks.len()); + let mut errs = Vec::with_capacity(disks.len()); + + for disk in disks.iter() { + let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); + futures.push(async move { + if let Some(disk) = disk { + match disk + .delete_version_with_namespace_owner( + bucket, + object, + fi.clone(), + force_del_marker, + DeleteOptions { + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + disk_namespace_owner, + ) + .await + { + Ok(r) => Ok(r), + Err(e) => Err(e), + } + } else { + Err(DiskError::DiskNotFound) + } + }); + } + + let results = join_all(futures).await; + for result in results { + match result { + Ok(_) => { + errs.push(None); + } + Err(e) => { + errs.push(Some(e)); + } + } + } + + let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object); + let should_rollback = quorum_result.is_err(); + let mut rollback_futures = Vec::new(); + for (index, err) in errs.iter().enumerate() { + // backlog#1158: when rolling back, fan the idempotent undo out to every + // online disk (each self-decides from its staged backup: restore if the + // rollback dir is present, no-op otherwise). This covers a disk that + // staged + applied the delete and *then* errored, which the plain + // `err.is_some()` skip would leave deleted while its peers were restored. + // On success only the successful disks' backup dirs need cleaning; errored + // disks' residue is reclaimed by heal/scanner. + if !should_rollback && err.is_some() { + continue; + } + + let Some(disk) = disks[index].as_ref() else { + continue; + }; + + let disk = disk.clone(); + let bucket = bucket.to_string(); + let object = object.to_string(); + let fi = fi.clone(); + let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); + rollback_futures.push(async move { + if should_rollback { + // The dedicated undo path never forwards the marker-only creation flag. + if let Err(err) = disk + .undo_write_with_namespace_owner( + &bucket, + &object, + fi, + DeleteOptions { + undo_write: true, + undo_delete: true, + old_data_dir: Some(rollback_dir), + ..Default::default() + }, + disk_namespace_owner, + ) + .await + { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to roll back delete after write quorum failure" + ); + } + } else { + let rollback_path = format!("{object}/{rollback_dir}"); + if let Err(err) = disk + .delete_with_namespace_owner( + &bucket, + &rollback_path, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + disk_namespace_owner, + ) + .await + && err != DiskError::FileNotFound + && err != DiskError::VolumeNotFound + { + warn!( + bucket = %bucket, + object = %object, + rollback_dir = %rollback_dir, + error = ?err, + "failed to clean delete rollback state after quorum success" + ); + } + } + }); + } + + join_all(rollback_futures).await; + drop(namespace_owner); + if quorum_result.is_ok() + && errs.iter().any(Option::is_some) + && let Some(purge) = delete_marker_purge + && let Some(version) = fi.version_id.filter(|version| !version.is_nil()) + { + let _ = self.persist_delete_marker_purge(bucket, object, version, purge).await; + } + // An explicit purge can carry deleted=true for the existing marker. + // It must not create a repair intent that could reintroduce that marker. + if quorum_result.is_ok() + && fi.deleted + && (fi.mark_deleted || force_del_marker) + && !fi.tier_free_version() + && version_purge_status_from_filemeta(fi.version_purge_status()) != VersionPurgeStatusType::Complete + && errs.iter().any(Option::is_some) + { + let version_id = fi.version_id.map(|version| version.to_string()); + let _ = self + .add_partial(bucket, object, version_id.as_deref().unwrap_or_default()) + .await; + } + quorum_result + } } #[async_trait::async_trait] @@ -7818,151 +8002,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } #[tracing::instrument(skip(self))] async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> { - let transported = delete_file_info_with_replication_transport_metadata(fi); - let fi = &transported; - let disks = self.disk_inventory().await; - let namespace_owner = (!is_meta_bucketname(bucket)).then(|| self.ctx.begin_namespace_commit()); - let write_quorum = disks.len() / 2 + 1; - let rollback_dir = Uuid::new_v4(); - - let mut futures = Vec::with_capacity(disks.len()); - let mut errs = Vec::with_capacity(disks.len()); - - for disk in disks.iter() { - let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); - futures.push(async move { - if let Some(disk) = disk { - match disk - .delete_version_with_namespace_owner( - bucket, - object, - fi.clone(), - force_del_marker, - DeleteOptions { - old_data_dir: Some(rollback_dir), - ..Default::default() - }, - disk_namespace_owner, - ) - .await - { - Ok(r) => Ok(r), - Err(e) => Err(e), - } - } else { - Err(DiskError::DiskNotFound) - } - }); - } - - let results = join_all(futures).await; - for result in results { - match result { - Ok(_) => { - errs.push(None); - } - Err(e) => { - errs.push(Some(e)); - } - } - } - - let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object); - let should_rollback = quorum_result.is_err(); - let mut rollback_futures = Vec::new(); - for (index, err) in errs.iter().enumerate() { - // backlog#1158: when rolling back, fan the idempotent undo out to every - // online disk (each self-decides from its staged backup: restore if the - // rollback dir is present, no-op otherwise). This covers a disk that - // staged + applied the delete and *then* errored, which the plain - // `err.is_some()` skip would leave deleted while its peers were restored. - // On success only the successful disks' backup dirs need cleaning; errored - // disks' residue is reclaimed by heal/scanner. - if !should_rollback && err.is_some() { - continue; - } - - let Some(disk) = disks[index].as_ref() else { - continue; - }; - - let disk = disk.clone(); - let bucket = bucket.to_string(); - let object = object.to_string(); - let fi = fi.clone(); - let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc); - rollback_futures.push(async move { - if should_rollback { - // The dedicated undo path never forwards the marker-only creation flag. - if let Err(err) = disk - .undo_write_with_namespace_owner( - &bucket, - &object, - fi, - DeleteOptions { - undo_write: true, - undo_delete: true, - old_data_dir: Some(rollback_dir), - ..Default::default() - }, - disk_namespace_owner, - ) - .await - { - warn!( - bucket = %bucket, - object = %object, - rollback_dir = %rollback_dir, - error = ?err, - "failed to roll back delete after write quorum failure" - ); - } - } else { - let rollback_path = format!("{object}/{rollback_dir}"); - if let Err(err) = disk - .delete_with_namespace_owner( - &bucket, - &rollback_path, - DeleteOptions { - recursive: true, - immediate: true, - ..Default::default() - }, - disk_namespace_owner, - ) - .await - && err != DiskError::FileNotFound - && err != DiskError::VolumeNotFound - { - warn!( - bucket = %bucket, - object = %object, - rollback_dir = %rollback_dir, - error = ?err, - "failed to clean delete rollback state after quorum success" - ); - } - } - }); - } - - join_all(rollback_futures).await; - drop(namespace_owner); - // An explicit purge can carry deleted=true for the existing marker. - // It must not create a repair intent that could reintroduce that marker. - if quorum_result.is_ok() - && fi.deleted - && (fi.mark_deleted || force_del_marker) - && !fi.tier_free_version() - && version_purge_status_from_filemeta(fi.version_purge_status()) != VersionPurgeStatusType::Complete - && errs.iter().any(Option::is_some) - { - let version_id = fi.version_id.map(|version| version.to_string()); - let _ = self - .add_partial(bucket, object, version_id.as_deref().unwrap_or_default()) - .await; - } - quorum_result + self.delete_object_version_with_purge(bucket, object, fi, force_del_marker, None) + .await } #[tracing::instrument(skip(self, objects, opts))] @@ -8081,6 +8122,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new(); let mut tier_reference_leases: Vec<(usize, String, Option)> = Vec::new(); let mut tier_free_version_receipt_candidates: HashMap = HashMap::new(); + let mut delete_marker_purge_candidates = vec![None; objects.len()]; for (i, dobj) in objects.iter().enumerate() { if del_errs[i].is_some() { @@ -8111,15 +8153,18 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let marker_delete = dobj.version_id.is_none() || dobj.synthetic_version_id; let replication_needs_source = replicate_delete && (!marker_delete || delete_config_snapshot.active_delete_marker_rules_require_tags(&replication_object_name)); - let (goi, gerr) = if object_lock_check_required + let (goi, authoritative_file_info, gerr) = if object_lock_check_required || replication_needs_source || opts.tier_delete_journal_api.is_some() || dobj.expected_identity.is_some() + || dobj.version_id.is_some() { - let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await; - (goi, gerr) + let (goi, file_info, _write_quorum, gerr) = self + .get_object_info_fileinfo_and_quorum(bucket, &dobj.object_name, &check_opts) + .await; + (goi, file_info, gerr) } else { - (ObjectInfo::default(), None) + (ObjectInfo::default(), FileInfo::default(), None) }; let source_missing = gerr .as_ref() @@ -8253,6 +8298,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // accounted as an object deletion (issue #6745). let removed_delete_marker = goi.delete_marker && dobj.version_id.is_some() && delete_file_info_version_id(goi.version_id) == version_id; + if removed_delete_marker + && goi.version_purge_status.is_empty() + && vr.version_purge_status().is_empty() + && vr.delete_marker_replication_status().is_empty() + && version_id.is_some_and(|version| authoritative_file_info.version_id == Some(version)) + { + delete_marker_purge_candidates[i] = + delete_marker_purge_candidate(&authoritative_file_info, opts.expected_bucket_incarnation_id); + } // Response-side only for the null identity: a delete request marked // `deleted` with `version_id == None` makes `FileMeta::delete_version` // re-create the marker it just removed (the suspended-bucket @@ -8616,6 +8670,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await; } + let purge_submissions = delete_marker_purge_candidates + .into_iter() + .enumerate() + .filter_map(|(index, purge)| { + let purge = purge?; + if del_errs[index].is_some() || !del_obj_errs.iter().any(|errors| errors[index].is_some()) { + return None; + } + let version = objects[index].version_id.filter(|version| !version.is_nil())?; + Some(self.persist_delete_marker_purge(bucket, objects[index].object_name.as_str(), version, purge)) + }); + let _ = join_all(purge_submissions).await; + for (object, err) in objects.iter().zip(del_errs.iter()) { if err.is_none() { self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await; @@ -8848,7 +8915,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let mut version_found = true; // delete_object_version below derives its own majority quorum from the // disk array, so the object-derived quorum here is unused. - let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await; + let (mut goi, authoritative_file_info, _write_quorum, gerr) = + self.get_object_info_fileinfo_and_quorum(bucket, object, &opts).await; if let Some(err) = &gerr && goi.name.is_empty() { @@ -8863,6 +8931,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { opts.precondition_check(&goi)?; check_object_lock_delete(&self.ctx, bucket, object, &goi, &opts).await?; } + let delete_marker_purge_candidate = opts + .version_id + .as_deref() + .and_then(|value| Uuid::parse_str(value).ok()) + .filter(|version| !version.is_nil() && authoritative_file_info.version_id == Some(*version)) + .and_then(|_| delete_marker_purge_candidate(&authoritative_file_info, opts.expected_bucket_incarnation_id)); if opts.transition.expire_restored { // Restore-expiry (DeleteRestoredAction / DeleteRestoredVersionAction) @@ -8921,6 +8995,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } let (mark_delete, mut delete_marker) = resolve_delete_version_state(&opts, &goi, version_found); + let delete_marker_purge = explicit_delete_removed_marker(&opts, &goi, version_found) + .then_some(delete_marker_purge_candidate) + .flatten(); let mod_time = if let Some(mt) = opts.mod_time { mt @@ -8969,9 +9046,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if opts.skip_free_version { fi.set_skip_tier_free_version(); } - self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts)) - .await - .map_err(|e| to_object_err(e, vec![bucket, object]))?; + self.delete_object_version_with_purge( + bucket, + object, + &fi, + should_force_delete_marker_for_missing_version(&opts), + delete_marker_purge.clone(), + ) + .await + .map_err(|e| to_object_err(e, vec![bucket, object]))?; #[cfg(test)] pause_delete_object_commit_after_publish(bucket, object).await; @@ -9021,7 +9104,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if opts.skip_free_version { dfi.set_skip_tier_free_version(); } - self.delete_object_version(bucket, object, &dfi, opts.delete_marker) + self.delete_object_version_with_purge(bucket, object, &dfi, opts.delete_marker, delete_marker_purge) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; #[cfg(test)] diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 30c9a5721..95d791218 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -625,26 +625,26 @@ impl SetDisks { } } - #[hotpath::measure(impl_type = "SetDisks")] - pub(super) async fn get_object_info_and_quorum( + pub(super) async fn get_object_info_fileinfo_and_quorum( &self, bucket: &str, object: &str, opts: &ObjectOptions, - ) -> (ObjectInfo, usize, Option) { + ) -> (ObjectInfo, FileInfo, usize, Option) { let snapshot = match self.get_object_fileinfo(bucket, object, opts, false, false).await { Ok(snapshot) => snapshot, - Err(e) => return (ObjectInfo::default(), 0, Some(e)), + Err(e) => return (ObjectInfo::default(), FileInfo::default(), 0, Some(e)), }; - let fi = snapshot.fi(); + let fi = snapshot.fi().clone(); let write_quorum = fi.write_quorum(self.default_write_quorum()); - let oi = ObjectInfo::from_file_info(fi, bucket, object, opts.versioned || opts.version_suspended); + let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); if !fi.version_purge_status().is_empty() && opts.version_id.is_some() { return ( oi, + fi, write_quorum, Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])), ); @@ -652,20 +652,26 @@ impl SetDisks { if fi.deleted { if opts.incl_free_versions && fi.tier_free_version() && opts.version_id.is_some() { - return (oi, write_quorum, None); + return (oi, fi, write_quorum, None); } return if opts.version_id.is_none() || opts.delete_marker { - (oi, write_quorum, Some(to_object_err(StorageError::FileNotFound, vec![bucket, object]))) + ( + oi, + fi, + write_quorum, + Some(to_object_err(StorageError::FileNotFound, vec![bucket, object])), + ) } else { ( oi, + fi, write_quorum, Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])), ) }; } - (oi, write_quorum, None) + (oi, fi, write_quorum, None) } #[allow(clippy::too_many_arguments)] @@ -2802,7 +2808,7 @@ mod metadata_cache_tests { } #[tokio::test] - async fn get_object_info_and_quorum_maps_delete_marker_and_purge_states() { + async fn get_object_info_fileinfo_and_quorum_maps_delete_marker_and_purge_states() { let bucket = "get-object-info-marker-bucket"; let (_dir, disk) = new_read_version_test_disk(bucket).await; let set = SetDisks::new( @@ -2829,8 +2835,8 @@ mod metadata_cache_tests { disk.write_metadata(bucket, bucket, "latest-delete-marker", latest_marker) .await .expect("latest marker metadata should be written"); - let (_, _, latest_err) = set - .get_object_info_and_quorum(bucket, "latest-delete-marker", &ObjectOptions::default()) + let (_, _, _, latest_err) = set + .get_object_info_fileinfo_and_quorum(bucket, "latest-delete-marker", &ObjectOptions::default()) .await; assert!( matches!(latest_err, Some(StorageError::ObjectNotFound(_, _))), @@ -2849,8 +2855,8 @@ mod metadata_cache_tests { disk.write_metadata(bucket, bucket, "version-delete-marker", version_marker) .await .expect("version marker metadata should be written"); - let (_, _, version_err) = set - .get_object_info_and_quorum( + let (_, _, _, version_err) = set + .get_object_info_fileinfo_and_quorum( bucket, "version-delete-marker", &ObjectOptions { diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index a7f04e64b..4c83e8395 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -19,6 +19,7 @@ use crate::services::rebalance::{REBAL_META_NAME, RebalStatus}; use crate::set_disk::get_lock_acquire_timeout; use crate::storage_api_contracts::heal::HealOperations as _; use crate::storage_api_contracts::namespace::NamespaceLocking as _; +use rustfs_common::mrf_channel::MrfDeleteMarkerPurge; use rustfs_lock::NamespaceLockGuard; use std::collections::BTreeSet; use tracing::trace; @@ -143,6 +144,72 @@ fn heal_format_fence_lost_error() -> Error { } impl ECStore { + pub async fn purge_delete_marker_with_proof( + self: &Arc, + bucket: &str, + object: &str, + version_id: &str, + purge: &MrfDeleteMarkerPurge, + opts: &HealOpts, + ) -> Result { + if opts.dry_run || opts.no_lock || !purge.is_valid() { + return Err(StorageError::PreconditionFailed); + } + let version = Uuid::parse_str(version_id) + .ok() + .filter(|version| !version.is_nil()) + .ok_or(StorageError::PreconditionFailed)?; + let (Some(pool_index), Some(set_index)) = (opts.pool, opts.set) else { + return Err(StorageError::PreconditionFailed); + }; + let mut marker = rustfs_filemeta::MetaDeleteMarker::default(); + let consumed = marker.unmarshal_msg(&purge.marker)?; + if usize::try_from(consumed).ok() != Some(purge.marker.len()) + || marker.version_id != Some(version) + || marker.stable_identity() != purge.marker_identity + { + return Err(StorageError::PreconditionFailed); + } + let marker_info = marker.clone().into_fileinfo(bucket, object, false)?; + if !marker_info.is_canonical_delete_marker() + || marker_info.delete_marker_incarnation() != Some(purge.marker_incarnation_id) + { + return Err(StorageError::PreconditionFailed); + } + + let current = self.bucket_incarnation_id_from_disk(bucket).await?; + if current.is_nil() { + return Err(StorageError::BucketNotFound(bucket.to_owned())); + } + let original = purge.bucket_incarnation_id; + let object = object.to_owned(); + self.run_bucket_heal_at_incarnation(bucket, current, opts, move |store, bucket, _opts| async move { + let scope = bucket_heal_scope(&bucket).ok_or(StorageError::PreconditionFailed)?; + scope.check()?; + if original != current { + let retirement_store = crate::bucket::metadata_sys::object_store_if_initialized_in(&store.ctx) + .await + .ok_or(StorageError::PreconditionFailed)?; + if !crate::bucket::retirement::is_retired(retirement_store, &bucket, original).await? { + return Err(StorageError::PreconditionFailed); + } + } + let pool = store + .pools + .get(pool_index) + .ok_or_else(|| invalid_heal_pool_index(pool_index, store.pools.len()))?; + let set = pool.disk_set.get(set_index).cloned().ok_or_else(|| { + StorageError::InvalidArgument( + "heal".to_string(), + "set".to_string(), + format!("invalid heal set index {set_index} for pool {pool_index}"), + ) + })?; + set.purge_delete_marker_exact(&bucket, &object, version, marker).await + }) + .await + } + pub(super) async fn run_bucket_heal_at_incarnation( self: &Arc, bucket: &str, diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index 0b3ca8b1d..87beb3d80 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -37,6 +37,7 @@ use rustfs_utils::http::{ insert_bytes, is_internal_key, remove_bytes, strip_internal_prefix, strip_internal_prefix_preserving_case, target_delete_marker_versions, }; +use sha2::{Digest as _, Sha256}; const MSGPACK_EXT8: u8 = 0xc7; const MSGPACK_EXT16: u8 = 0xc8; @@ -3010,6 +3011,40 @@ impl TryFrom for MetaDeleteMarker { } impl MetaDeleteMarker { + /// Return a deterministic identity for an exact delete-marker body. + /// MessagePack map order is intentionally excluded from the identity. + pub fn stable_identity(&self) -> [u8; 32] { + const DOMAIN: &[u8] = b"rustfs-delete-marker-identity-v1\0"; + + let mut hasher = Sha256::new(); + hasher.update(DOMAIN); + match self.version_id { + Some(version_id) => { + hasher.update([1]); + hasher.update(version_id.as_bytes()); + } + None => hasher.update([0]), + } + match self.mod_time { + Some(mod_time) => { + hasher.update([1]); + hasher.update(mod_time.unix_timestamp_nanos().to_be_bytes()); + } + None => hasher.update([0]), + } + + let mut metadata = self.meta_sys.iter().collect::>(); + metadata.sort_unstable_by(|(left, _), (right, _)| left.as_bytes().cmp(right.as_bytes())); + hasher.update(u64::try_from(metadata.len()).unwrap_or(u64::MAX).to_be_bytes()); + for (key, value) in metadata { + hasher.update(u64::try_from(key.len()).unwrap_or(u64::MAX).to_be_bytes()); + hasher.update(key.as_bytes()); + hasher.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes()); + hasher.update(value); + } + hasher.finalize().into() + } + pub fn free_version(&self) -> bool { contains_key_bytes(&self.meta_sys, SUFFIX_FREE_VERSION) } @@ -5625,6 +5660,40 @@ mod tests { assert_eq!(base.get_signature(), base.get_signature()); } + #[test] + fn delete_marker_stable_identity_is_order_independent_and_exact() { + let version_id = sample_version_id(); + let mod_time = sample_mod_time(); + let mut first = MetaDeleteMarker { + version_id: Some(version_id), + mod_time: Some(mod_time), + meta_sys: HashMap::new(), + }; + first.meta_sys.insert("alpha".to_string(), vec![1, 2]); + first.meta_sys.insert("beta".to_string(), vec![3, 4]); + + let mut reordered = MetaDeleteMarker { + version_id: Some(version_id), + mod_time: Some(mod_time), + meta_sys: HashMap::new(), + }; + reordered.meta_sys.insert("beta".to_string(), vec![3, 4]); + reordered.meta_sys.insert("alpha".to_string(), vec![1, 2]); + assert_eq!(first.stable_identity(), reordered.stable_identity()); + + let mut changed = first.clone(); + changed.meta_sys.insert("beta".to_string(), vec![3, 5]); + assert_ne!(first.stable_identity(), changed.stable_identity()); + + changed = first.clone(); + changed.version_id = Some(Uuid::new_v4()); + assert_ne!(first.stable_identity(), changed.stable_identity()); + + changed = first.clone(); + changed.mod_time = changed.mod_time.map(|value| value + time::Duration::NANOSECOND); + assert_ne!(first.stable_identity(), changed.stable_identity()); + } + #[test] fn invalid_version_uses_error_sentinel_not_zero() { // A version whose inner body is missing must not masquerade as an diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 88cbe3389..0cbd06c7c 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -134,6 +134,7 @@ pub(super) struct MrfRepairNoticeTarget { pub(super) version_id: Option<[u8; 16]>, pub(super) kind: rustfs_common::mrf_channel::MrfKind, pub(super) scope: Option, + pub(super) delete_marker_purge: Option, pub(super) lease: Option, } @@ -493,7 +494,8 @@ fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { HealType::Cluster => false, HealType::Object { bucket, object, .. } | HealType::Metadata { bucket, object } - | HealType::ECDecode { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object), + | HealType::ECDecode { bucket, object, .. } + | HealType::DeleteMarkerPurge { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object), HealType::Bucket { bucket } => heal_path == bucket, HealType::Prefix { bucket, prefix } => heal_path_matches_bucket_child(heal_path, bucket, prefix), HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id, @@ -629,7 +631,8 @@ fn heal_type_path_view(heal_type: &HealType) -> Option<(&str, &str, bool)> { HealType::Prefix { bucket, prefix } => Some((bucket, prefix, true)), HealType::Object { bucket, object, .. } | HealType::Metadata { bucket, object } - | HealType::ECDecode { bucket, object, .. } => Some((bucket, object, false)), + | HealType::ECDecode { bucket, object, .. } + | HealType::DeleteMarkerPurge { bucket, object, .. } => Some((bucket, object, false)), HealType::Cluster | HealType::ErasureSet { .. } => None, } } @@ -1754,6 +1757,7 @@ impl HealManager { let kind = match &request.heal_type { HealType::Metadata { .. } => rustfs_common::mrf_channel::MrfKind::MetadataCorruption, HealType::ECDecode { .. } => rustfs_common::mrf_channel::MrfKind::DecodeFailure, + HealType::DeleteMarkerPurge { .. } => rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge, _ => rustfs_common::mrf_channel::MrfKind::PartialWrite, }; self.submit_mrf_heal_request_with_receipt_and_identity( @@ -1764,6 +1768,7 @@ impl HealManager { version_id, kind, scope: None, + delete_marker_purge: None, lease: None, }, ) @@ -1775,7 +1780,7 @@ impl HealManager { request: HealRequest, mrf_notice_target: MrfRepairNoticeTarget, ) -> Result { - let preserve_alias = mrf_notice_target.kind != rustfs_common::mrf_channel::MrfKind::PartialWrite; + let preserve_alias = !mrf_notice_target.kind.is_durable(); self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target)) .await } @@ -1784,6 +1789,9 @@ impl HealManager { &self, intent: &rustfs_common::mrf_channel::MrfIntent, ) -> Option { + if let Some(purge) = intent.delete_marker_purge.as_ref() { + return rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, purge.bucket_incarnation_id); + } 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) @@ -2030,10 +2038,7 @@ impl HealManager { // of the silent merge (MinIO's ErrHealAlreadyRunning). let admission = if (request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError) - || (duplicate_state != "queued" - && mrf_notice_target - .as_ref() - .is_some_and(|target| target.kind == rustfs_common::mrf_channel::MrfKind::PartialWrite)) + || (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable())) { // A running/retrying task may have observed the object before // this write committed. Its receipt cannot prove the new lease. diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index 2607f4256..721d93a36 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -126,6 +126,17 @@ impl CompletedHealStatus { add(object.capacity()); add(version_id.as_ref().map_or(0, String::capacity)); } + HealType::DeleteMarkerPurge { + bucket, + object, + version_id, + purge, + } => { + add(bucket.capacity()); + add(object.capacity()); + add(version_id.capacity()); + add(purge.marker.len()); + } HealType::Prefix { bucket, prefix } => { add(bucket.capacity()); add(prefix.capacity()); @@ -490,6 +501,17 @@ impl PriorityHealQueue { } => { format!("ecdecode:{}:{}:{}", bucket, object, version_id.as_deref().unwrap_or("")) } + HealType::DeleteMarkerPurge { + bucket, + object, + version_id, + purge, + } => format!( + "delete-marker-purge:{bucket}:{object}:{version_id}:{}:{}:{}", + purge.bucket_incarnation_id, + purge.marker_incarnation_id, + base64_simd::URL_SAFE_NO_PAD.encode_to_string(purge.marker_identity) + ), } } diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index 19dca0d64..5eb51df8e 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -149,9 +149,11 @@ impl RecoveryHealType { } } -impl From<&HealType> for RecoveryHealType { - fn from(heal_type: &HealType) -> Self { - match heal_type { +impl TryFrom<&HealType> for RecoveryHealType { + type Error = Error; + + fn try_from(heal_type: &HealType) -> Result { + Ok(match heal_type { HealType::Cluster => Self::Cluster, HealType::Bucket { bucket } => Self::Bucket { bucket: bucket.clone() }, HealType::Object { @@ -185,7 +187,10 @@ impl From<&HealType> for RecoveryHealType { object: object.clone(), version_id: version_id.clone(), }, - } + HealType::DeleteMarkerPurge { .. } => { + return Err(Error::Other("Delete-marker purge recovery is owned by the MRF journal".to_string())); + } + }) } } @@ -309,28 +314,28 @@ impl RootHealTerminal { Ok(()) } - fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Self { - Self { + fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Result { + Ok(Self { schema: ROOT_TERMINAL_SCHEMA, task_id: task_id.to_owned(), - heal_type: RecoveryHealType::from(&completed.heal_type), + heal_type: RecoveryHealType::try_from(&completed.heal_type)?, status: completed.status.clone(), options: completed.options.clone(), progress: completed.progress.clone(), completed_at: completed.completed_at, - } + }) } - fn cancelled(task_id: &str, heal_type: &HealType, options: HealOptions) -> Self { - Self { + fn cancelled(task_id: &str, heal_type: &HealType, options: HealOptions) -> Result { + Ok(Self { schema: ROOT_TERMINAL_SCHEMA, task_id: task_id.to_owned(), - heal_type: RecoveryHealType::from(heal_type), + heal_type: RecoveryHealType::try_from(heal_type)?, status: HealTaskStatus::Cancelled, options, progress: None, completed_at: SystemTime::now(), - } + }) } fn into_completed(self) -> CompletedHealStatus { @@ -357,17 +362,17 @@ impl RootHealTerminal { } impl RootHealIntent { - fn from_request(request: &HealRequest) -> Self { - Self { + fn from_request(request: &HealRequest) -> Result { + Ok(Self { schema: ROOT_RECOVERY_SCHEMA, task_id: request.id.clone(), - heal_type: RecoveryHealType::from(&request.heal_type), + heal_type: RecoveryHealType::try_from(&request.heal_type)?, bucket_incarnation_id: request.bucket_incarnation_id, options: request.options.clone(), priority: request.priority, retry_attempts: request.retry_attempts, created_at: request.created_at, - } + }) } fn into_request(self) -> HealRequest { @@ -919,7 +924,7 @@ impl RootHealRecovery { if request.options.no_lock { return Err(Error::Other("Administrator root heal cannot skip namespace locking".to_string())); } - let intent = RootHealIntent::from_request(request); + let intent = RootHealIntent::from_request(request)?; intent.heal_type.validate()?; let bytes = serde_json::to_vec(&intent).map_err(|error| Error::Other(format!("Serialize root heal recovery record: {error}")))?; @@ -1076,7 +1081,7 @@ impl RootHealRecovery { }; let pending = decode_intent(task_id, &bytes)?; let heal_type = HealType::from(pending.heal_type); - let terminal = RootHealTerminal::cancelled(task_id, &heal_type, pending.options); + let terminal = RootHealTerminal::cancelled(task_id, &heal_type, pending.options)?; let completed = terminal.clone().into_completed(); let _ = Self::persist_terminal_locked(&disks, task_id, terminal, &completed).await?; match EcstoreDiskAPI::compare_and_update_file( @@ -1110,7 +1115,7 @@ impl RootHealRecovery { let _guard = self.mutation.lock().await; let disks = self.disks().await?; let pending = - Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed), completed) + Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed)?, completed) .await?; #[cfg(test)] if self @@ -1373,7 +1378,7 @@ impl HealManager { { Ok(()) => {} Err(error @ Error::StaleBucketIncarnation { .. }) => { - let mut terminal = RootHealTerminal::cancelled(&request.id, &request.heal_type, request.options.clone()); + let mut terminal = RootHealTerminal::cancelled(&request.id, &request.heal_type, request.options.clone())?; terminal.status = HealTaskStatus::Failed { error: error.to_string(), }; diff --git a/crates/heal/src/heal/manager/root_recovery/report.rs b/crates/heal/src/heal/manager/root_recovery/report.rs index 22d0e711a..923daf67b 100644 --- a/crates/heal/src/heal/manager/root_recovery/report.rs +++ b/crates/heal/src/heal/manager/root_recovery/report.rs @@ -203,7 +203,8 @@ mod tests { #[test] fn root_report_encoding_rejects_oversized_payload() { let task_id = Uuid::new_v4().to_string(); - let terminal = RootHealTerminal::cancelled(&task_id, &HealType::Cluster, HealOptions::default()); + let terminal = RootHealTerminal::cancelled(&task_id, &HealType::Cluster, HealOptions::default()) + .expect("cluster is an administrator recovery type"); let mut completed = terminal.clone().into_completed(); completed.progress = Some(HealProgress { current_object: Some("x".repeat(MAX_ROOT_REPORT_BYTES)), diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 20f8c261b..10da79a38 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -818,6 +818,7 @@ pub(super) fn mrf_verified_repair_event_for_target( MrfKind::DecodeFailure => HealObjectKind::Decode, MrfKind::MetadataCorruption => HealObjectKind::Metadata, MrfKind::PartialWrite => HealObjectKind::Object, + MrfKind::DeleteMarkerPurge => HealObjectKind::DeleteMarkerPurge, }; if outcome.identity.kind != expected_kind || outcome.identity.bucket.as_str() != target.bucket.as_ref() @@ -848,6 +849,7 @@ pub(super) fn mrf_verified_repair_event_for_target( object: target.object.clone(), version_id, scope, + delete_marker_purge: target.delete_marker_purge, 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 ea1099b0b..f6e26f45f 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -1226,6 +1226,7 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() { pool_index: 1, set_index: 2, }), + delete_marker_purge: None, lease: None, }; let matching = HealObjectOutcome { @@ -1352,6 +1353,7 @@ fn completed_mrf_notice_publishes_only_verified_positive_events() { pool_index: 1, set_index: 2, }), + delete_marker_purge: None, lease: None, }; let mismatch_target = MrfRepairNoticeTarget { diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 2dfe4e09a..46018012f 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -26,8 +26,8 @@ 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, MrfDurableAdmissionError, MrfDurableRepairAnchor, MrfDurableSubmission, MrfIngressResult, MrfIntent, - MrfKind, + MRF_MAX_ATTEMPTS, MRF_MAX_DELETE_MARKER_BYTES, MrfDeleteMarkerPurge, MrfDeleteMarkerPurgeIdentity, MrfDurableAdmissionError, + MrfDurableRepairAnchor, MrfDurableSubmission, MrfIngressResult, MrfIntent, MrfKind, }; use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; use std::collections::{HashSet, VecDeque}; @@ -48,7 +48,7 @@ pub mod snapshot; /// layout. pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin"; /// The scoped path is the authoritative snapshot for new readers and carries -/// both v1 and v2 records. The legacy path is only a v1 compatibility mirror; +/// v1, v2, and v3 records. The legacy path is only a v1 compatibility mirror; /// older readers ignore the authoritative path, while new readers never merge /// the two files. This prevents a partial two-file flush from fabricating a /// mixed epoch. @@ -59,6 +59,7 @@ const MRF_JOURNAL_FORMAT: u8 = 1; /// Record layout version. const MRF_JOURNAL_VERSION: u8 = 1; const MRF_JOURNAL_VERSION_SCOPED: u8 = 2; +const MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE: u8 = 3; /// Fixed header size: format, version, kind, attempts, enqueued_at_ms, /// has_version flag. @@ -125,6 +126,7 @@ struct MrfQueueKey { object: Arc, version_id: Option<[u8; 16]>, scope: Option, + delete_marker_purge: Option, } fn queue_key(intent: &MrfIntent) -> MrfQueueKey { @@ -138,6 +140,7 @@ fn queue_key(intent: &MrfIntent) -> MrfQueueKey { object: intent.object.clone(), version_id, scope, + delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity), } } @@ -234,9 +237,19 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) -> bool { .then_some(intent.scope) .flatten(); let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]); + let purge = match intent.kind { + MrfKind::DeleteMarkerPurge => match intent.delete_marker_purge.as_ref().filter(|payload| payload.is_valid()) { + Some(payload) if version_id.is_some() && scope.is_some() => Some(payload), + _ => return false, + }, + _ if intent.delete_marker_purge.is_none() => None, + _ => return false, + }; let start = out.len(); out.push(MRF_JOURNAL_FORMAT); - out.push(if scope.is_some() { + out.push(if purge.is_some() { + MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE + } else if scope.is_some() { MRF_JOURNAL_VERSION_SCOPED } else { MRF_JOURNAL_VERSION @@ -245,6 +258,7 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) -> bool { rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1, rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2, rustfs_common::mrf_channel::MrfKind::PartialWrite => 3, + rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge => 4, }); out.push(intent.attempts); out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes()); @@ -263,6 +277,17 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) -> bool { out.extend_from_slice(&object_len.to_le_bytes()); out.extend_from_slice(intent.bucket.as_bytes()); out.extend_from_slice(intent.object.as_bytes()); + if let Some(purge) = purge { + let Ok(marker_len) = u32::try_from(purge.marker.len()) else { + out.truncate(start); + return false; + }; + out.extend_from_slice(purge.bucket_incarnation_id.as_bytes()); + out.extend_from_slice(purge.marker_incarnation_id.as_bytes()); + out.extend_from_slice(&purge.marker_identity); + out.extend_from_slice(&marker_len.to_le_bytes()); + out.extend_from_slice(&purge.marker); + } let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); hasher.update(&out[start..]); let Ok(checksum) = u32::try_from(hasher.finalize()) else { @@ -277,15 +302,24 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { if data.len() < MRF_RECORD_FIXED_HEAD + 8 { return None; } - if data[0] != MRF_JOURNAL_FORMAT || !matches!(data[1], MRF_JOURNAL_VERSION | MRF_JOURNAL_VERSION_SCOPED) { + if data[0] != MRF_JOURNAL_FORMAT + || !matches!( + data[1], + MRF_JOURNAL_VERSION | MRF_JOURNAL_VERSION_SCOPED | MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE + ) + { return None; } let kind = match data[2] { 1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure, 2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption, 3 => rustfs_common::mrf_channel::MrfKind::PartialWrite, + 4 => rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge, _ => return None, }; + if (kind == MrfKind::DeleteMarkerPurge) != (data[1] == MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE) { + return None; + } let attempts = data[3]; let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?); let has_version = match data[12] { @@ -304,7 +338,10 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { } else { None }; - let scope = if data[1] == MRF_JOURNAL_VERSION_SCOPED { + if kind == MrfKind::DeleteMarkerPurge && version_id.is_none_or(|bytes| bytes == [0; 16]) { + return None; + } + let scope = if matches!(data[1], MRF_JOURNAL_VERSION_SCOPED | MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE) { if data.len() < cursor + 8 { return None; } @@ -324,7 +361,43 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { return None; } cursor += 8; - let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?; + let identity_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?; + if data.len() < identity_end { + return None; + } + let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?); + let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..identity_end]).ok()?); + let (delete_marker_purge, body_end) = if kind == MrfKind::DeleteMarkerPurge { + const PURGE_FIXED: usize = 16 + 16 + 32 + 4; + if data.len() < identity_end.checked_add(PURGE_FIXED)? { + return None; + } + let mut purge_cursor = identity_end; + let bucket_incarnation_id = Uuid::from_slice(&data[purge_cursor..purge_cursor + 16]).ok()?; + purge_cursor += 16; + let marker_incarnation_id = Uuid::from_slice(&data[purge_cursor..purge_cursor + 16]).ok()?; + purge_cursor += 16; + let marker_identity = data[purge_cursor..purge_cursor + 32].try_into().ok()?; + purge_cursor += 32; + let marker_len = usize::try_from(u32::from_le_bytes(data[purge_cursor..purge_cursor + 4].try_into().ok()?)).ok()?; + purge_cursor += 4; + if marker_len == 0 || marker_len > MRF_MAX_DELETE_MARKER_BYTES { + return None; + } + let body_end = purge_cursor.checked_add(marker_len)?; + if data.len() < body_end { + return None; + } + let payload = MrfDeleteMarkerPurge::new( + bucket_incarnation_id, + marker_incarnation_id, + marker_identity, + data[purge_cursor..body_end].to_vec(), + )?; + (Some(payload), body_end) + } else { + (None, identity_end) + }; let record_end = body_end.checked_add(4)?; if data.len() < record_end { return None; @@ -334,14 +407,13 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { if u32::try_from(hasher.finalize()).ok()? != u32::from_le_bytes(data[body_end..record_end].try_into().ok()?) { return None; } - let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?); - let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?); Some(( MrfIntent { bucket, object, version_id, kind, + delete_marker_purge, scope: if matches!(kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) { None } else { @@ -482,6 +554,18 @@ pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest { }, HealPriority::Normal, ), + rustfs_common::mrf_channel::MrfKind::DeleteMarkerPurge => ( + HealType::DeleteMarkerPurge { + bucket, + object, + version_id: version_id.expect("validated delete-marker purge requires a version"), + purge: intent + .delete_marker_purge + .clone() + .expect("validated delete-marker purge requires a payload"), + }, + HealPriority::Normal, + ), }; let mut options = HealOptions::default(); if matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::PartialWrite) { @@ -510,6 +594,7 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c version_id: intent.version_id, kind: intent.kind, scope: intent.scope, + delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity), lease: intent.lease, }, ) @@ -536,7 +621,7 @@ struct MrfRuntime { /// 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 + /// Durable responsibilities accepted from replay and waiting for an /// exact storage-owned proof before the startup journal can be deleted. durable_replay_anchors: Vec, /// Startup replay source to remove after the retained replay @@ -568,7 +653,7 @@ impl MrfRuntime { // later write replaces this lease before its old proof arrives. let adopted_leases: HashSet<_> = self.partial_writes.intents().filter_map(|intent| intent.lease).collect(); self.durable_replay_anchors - .retain(|anchor| anchor.kind != MrfKind::PartialWrite || !adopted_leases.contains(&anchor.lease)); + .retain(|anchor| !anchor.kind.is_durable() || !adopted_leases.contains(&anchor.lease)); } fn admit_partial_write(&mut self, intent: MrfIntent) -> Result<(), MrfDurableAdmissionError> { @@ -664,7 +749,7 @@ impl MrfRuntime { self.backoff_until = None; } while let Some(mut intent) = self.queue.pop_front() { - if intent.kind == MrfKind::PartialWrite { + if intent.kind.is_durable() { if self.admit_partial_write(intent.clone()).is_err() { self.queue.push_back(intent); break; @@ -780,9 +865,10 @@ impl MrfRuntime { } } -/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and -/// spawn the consumer task. Called once from the heal runtime bootstrap right -/// after the manager started; a disabled feature or a double call is a no-op. +/// Initialize the global MRF channels and spawn the consumer task. The +/// `RUSTFS_HEAL_MRF_ENABLE` switch only disables best-effort ingress; durable +/// responsibility replay remains active so committed marker purges survive the +/// ordinary MRF kill-switch. /// Public for integration tests that drive the real consumer loop. pub fn spawn_mrf_consumer(manager: Arc) { let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE); @@ -790,9 +876,8 @@ pub fn spawn_mrf_consumer(manager: Arc) { if !enabled { tracing::info!( target: "rustfs::heal::mrf", - "MRF intent pipeline disabled by configuration; producers will not deliver" + "Best-effort MRF ingress disabled by configuration; durable responsibilities remain active" ); - return; } let (receiver, durable_receiver) = match rustfs_common::mrf_channel::init_mrf_channel().and_then(|receiver| { rustfs_common::mrf_channel::init_durable_mrf_channel().map(|durable_receiver| (receiver, durable_receiver)) @@ -995,7 +1080,7 @@ async fn replay_into( *backoff_until = Some(tokio::time::Instant::now()); break; } - if intent.kind == MrfKind::PartialWrite { + if intent.kind.is_durable() { // Preserve the executable record as well as its proof anchor: // a target that is still offline during replay needs live retries. if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await { @@ -1146,7 +1231,7 @@ async fn run_mrf_consumer( return; } for intent in batch.drain(..) { - if intent.kind == MrfKind::PartialWrite { + if intent.kind.is_durable() { if runtime.admit_partial_write(intent.clone()).is_err() { rustfs_common::mrf_channel::release_mrf_intent(&intent); counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1); @@ -1273,6 +1358,7 @@ mod tests { object: StdArc::from(object), version_id: Some([7u8; 16]), kind: MrfKind::DecodeFailure, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 1_700_000_000_000, @@ -1280,6 +1366,38 @@ mod tests { } } + fn purge_intent(marker: Vec) -> MrfIntent { + let bucket_incarnation_id = Uuid::new_v4(); + MrfIntent { + bucket: StdArc::from("purge-bucket"), + object: StdArc::from("markers/object.bin"), + version_id: Some(*Uuid::new_v4().as_bytes()), + kind: MrfKind::DeleteMarkerPurge, + delete_marker_purge: MrfDeleteMarkerPurge::new(bucket_incarnation_id, bucket_incarnation_id, [9; 32], marker), + scope: Some(rustfs_common::mrf_channel::MrfScope { + pool_index: 2, + set_index: 3, + }), + lease: None, + enqueued_at_ms: 1_700_000_000_001, + attempts: 2, + } + } + + fn rewrite_record_crc(record: &mut [u8]) { + let crc_offset = record.len() - 4; + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&record[..crc_offset]); + let checksum = u32::try_from(hasher.finalize()).expect("CRC32 fits"); + record[crc_offset..].copy_from_slice(&checksum.to_le_bytes()); + } + + fn assert_journal_record_rejected(record: &[u8]) { + let (decoded, truncated) = decode_journal(record); + assert!(decoded.is_empty()); + assert_eq!(truncated, record.len()); + } + fn encoded_payload(intent: &MrfIntent) -> Vec { let mut payload = Vec::new(); assert!(encode_intent(intent, &mut payload), "fixture intent must encode"); @@ -1432,6 +1550,7 @@ mod tests { object: anchor.object.clone(), version_id: anchor.version_id, scope: anchor.scope, + delete_marker_purge: anchor.delete_marker_purge, lease: Some(anchor.lease), bucket_incarnation_id: anchor.bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -2113,6 +2232,7 @@ mod tests { object: intent.object.clone(), version_id: intent.version_id, scope: intent.scope, + delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity), lease: intent.lease, bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -2368,6 +2488,7 @@ mod tests { object: anchor.object.clone(), version_id: anchor.version_id, scope: anchor.scope, + delete_marker_purge: anchor.delete_marker_purge, lease: Some(anchor.lease), bucket_incarnation_id: anchor.bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::Repaired, @@ -2629,6 +2750,7 @@ mod tests { object: StdArc::from("object/c"), version_id: None, kind: MrfKind::MetadataCorruption, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 5, @@ -2651,6 +2773,76 @@ mod tests { } } + #[test] + fn journal_v3_roundtrip_preserves_delete_marker_purge_payload() { + let intent = purge_intent(vec![1, 2, 3, 4, 5]); + let mut buf = Vec::new(); + assert!(encode_intent(&intent, &mut buf)); + assert_eq!(buf[1], MRF_JOURNAL_VERSION_DELETE_MARKER_PURGE); + assert_eq!(buf[2], 4); + + let (decoded, truncated) = decode_journal(&buf); + assert_eq!(truncated, 0); + assert_eq!(decoded.len(), 1); + let decoded = &decoded[0]; + assert_eq!(decoded.bucket, intent.bucket); + assert_eq!(decoded.object, intent.object); + assert_eq!(decoded.version_id, intent.version_id); + assert_eq!(decoded.kind, MrfKind::DeleteMarkerPurge); + assert_eq!(decoded.scope, intent.scope); + assert_eq!(decoded.delete_marker_purge, intent.delete_marker_purge); + assert_eq!(decoded.attempts, intent.attempts); + } + + #[test] + fn journal_v3_rejects_nil_version_wrong_layout_corruption_and_truncation() { + let intent = purge_intent(vec![1, 2, 3, 4, 5]); + let mut encoded = Vec::new(); + assert!(encode_intent(&intent, &mut encoded)); + + let mut nil_version = encoded.clone(); + nil_version[13..29].fill(0); + rewrite_record_crc(&mut nil_version); + assert_journal_record_rejected(&nil_version); + + let mut wrong_kind = encoded.clone(); + wrong_kind[2] = 3; + rewrite_record_crc(&mut wrong_kind); + assert_journal_record_rejected(&wrong_kind); + + let mut wrong_layout = encoded.clone(); + wrong_layout[1] = MRF_JOURNAL_VERSION_SCOPED; + rewrite_record_crc(&mut wrong_layout); + assert_journal_record_rejected(&wrong_layout); + + let marker_byte = encoded.len() - 5; + encoded[marker_byte] ^= 0xff; + assert_journal_record_rejected(&encoded); + + let torn = &encoded[..encoded.len() - 1]; + assert_journal_record_rejected(torn); + } + + #[test] + fn journal_v3_rejects_missing_or_oversized_payload_without_mutating_output() { + let mut missing = purge_intent(vec![1]); + missing.delete_marker_purge = None; + let mut output = vec![7]; + assert!(!encode_intent(&missing, &mut output)); + assert_eq!(output, vec![7]); + + let bucket_incarnation_id = Uuid::new_v4(); + let mut oversized = purge_intent(vec![1]); + oversized.delete_marker_purge = Some(MrfDeleteMarkerPurge { + bucket_incarnation_id, + marker_incarnation_id: bucket_incarnation_id, + marker_identity: [3; 32], + marker: StdArc::from(vec![0; MRF_MAX_DELETE_MARKER_BYTES + 1]), + }); + assert!(!encode_intent(&oversized, &mut output)); + assert_eq!(output, vec![7]); + } + #[test] fn journal_torn_tail_is_truncated() { let mut buf = Vec::new(); @@ -2701,6 +2893,7 @@ mod tests { object: StdArc::from("o"), version_id: None, kind: MrfKind::MetadataCorruption, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 0, @@ -2714,6 +2907,7 @@ mod tests { object: StdArc::from("o"), version_id: None, kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 0, diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index f17eb456f..b53f73e06 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -24,7 +24,7 @@ struct Responsibility { next_attempt: Instant, } -/// Partial writes have no rediscovery producer when scanning is disabled. +/// Durable storage responsibilities have no guaranteed rediscovery producer. /// Admission, task failure and retry exhaustion cannot release their records. #[derive(Default)] pub(super) struct PartialWrites { @@ -169,6 +169,7 @@ mod tests { object: Arc::from(object), version_id: None, kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: Some(MrfScope { pool_index: 0, set_index: 0, diff --git a/crates/heal/src/heal/mrf_queue/snapshot.rs b/crates/heal/src/heal/mrf_queue/snapshot.rs index 6164bd2c3..a5514df5e 100644 --- a/crates/heal/src/heal/mrf_queue/snapshot.rs +++ b/crates/heal/src/heal/mrf_queue/snapshot.rs @@ -701,6 +701,7 @@ mod tests { object: Arc::from(object), version_id: None, kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 1234, @@ -719,6 +720,7 @@ mod tests { object: Arc::from(format!("o-{index:06}")), version_id: None, kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: None, lease: None, enqueued_at_ms: 1234, @@ -1699,6 +1701,7 @@ mod tests { object: Arc::from("a"), version_id: None, kind: MrfKind::PartialWrite, + delete_marker_purge: None, scope: Some(MrfScope { pool_index: 0, set_index, diff --git a/crates/heal/src/heal/mrf_queue/snapshot/migration.rs b/crates/heal/src/heal/mrf_queue/snapshot/migration.rs index 97dedaa59..b46e59667 100644 --- a/crates/heal/src/heal/mrf_queue/snapshot/migration.rs +++ b/crates/heal/src/heal/mrf_queue/snapshot/migration.rs @@ -584,6 +584,7 @@ mod tests { object: Arc::from(object), version_id: None, kind, + delete_marker_purge: None, scope, lease: None, enqueued_at_ms: 1, diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index 1ad85fa5a..0a24bee47 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -29,6 +29,7 @@ pub enum HealObjectKind { Object, Metadata, Decode, + DeleteMarkerPurge, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 741314afc..955915348 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -15,6 +15,7 @@ use crate::{Error, Result}; use async_trait::async_trait; use base64_simd::URL_SAFE_NO_PAD; +use rustfs_common::mrf_channel::MrfDeleteMarkerPurge; use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealScanMode}; use rustfs_madmin::heal_commands::HealResultItem; use serde::{Deserialize, Serialize}; @@ -401,6 +402,18 @@ pub struct HealListItem { /// Heal storage layer interface #[async_trait] pub trait HealStorageAPI: Send + Sync { + /// Delete one exact marker from every member of its original erasure set. + async fn purge_delete_marker( + &self, + _bucket: &str, + _object: &str, + _version_id: &str, + _purge: &MrfDeleteMarkerPurge, + _opts: &HealOpts, + ) -> Result { + Err(Error::other("delete-marker purge is unsupported by this heal storage backend")) + } + /// Get object meta /// /// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): MRF intents @@ -846,6 +859,41 @@ fn is_transient_object_exists_error(err: &StorageError) -> bool { #[async_trait] impl HealStorageAPI for ECStoreHealStorage { + async fn purge_delete_marker( + &self, + bucket: &str, + object: &str, + version_id: &str, + purge: &MrfDeleteMarkerPurge, + opts: &HealOpts, + ) -> Result { + let removed = self + .ecstore + .purge_delete_marker_with_proof(bucket, object, version_id, purge, opts) + .await + .map_err(Error::Storage)?; + Ok(HealStorageObjectResult { + item: HealResultItem::default(), + error: None, + receipt: Some(HealObjectReceipt { + identity: HealObjectIdentity { + kind: HealObjectKind::DeleteMarkerPurge, + bucket: bucket.to_owned(), + object: object.to_owned(), + version_id: Some(version_id.to_owned()), + bucket_incarnation_id: Some(purge.bucket_incarnation_id), + pool_index: opts.pool, + set_index: opts.set, + }, + disposition: if removed { + HealObjectDisposition::Repaired + } else { + HealObjectDisposition::AuthoritativelyAbsent + }, + }), + }) + } + async fn admit_bucket_incarnation(&self, bucket: &str) -> Result { match self.ecstore.bucket_incarnation_id_from_disk(bucket).await { Ok(id) if !id.is_nil() => Ok(id), diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 1e0932bdb..87f038e30 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -130,6 +130,13 @@ pub enum HealType { object: String, version_id: Option, }, + /// Complete an explicit delete-marker purge on its original erasure set. + DeleteMarkerPurge { + bucket: String, + object: String, + version_id: String, + purge: rustfs_common::mrf_channel::MrfDeleteMarkerPurge, + }, } impl HealType { @@ -142,6 +149,7 @@ impl HealType { Self::ErasureSet { .. } => "erasure_set", Self::Metadata { .. } => "metadata", Self::ECDecode { .. } => "ec_decode", + Self::DeleteMarkerPurge { .. } => "delete_marker_purge", } } @@ -152,7 +160,10 @@ impl HealType { /// cannot amplify into per-object `info!`/`warn!` lines; aggregate kinds /// (cluster/bucket/prefix/erasure-set) keep operator-visible levels. pub(crate) fn is_per_object(&self) -> bool { - matches!(self, Self::Object { .. } | Self::Metadata { .. } | Self::ECDecode { .. }) + matches!( + self, + Self::Object { .. } | Self::Metadata { .. } | Self::ECDecode { .. } | Self::DeleteMarkerPurge { .. } + ) } } @@ -598,6 +609,7 @@ impl HealTask { kind: match self.heal_type { HealType::Metadata { .. } => HealObjectKind::Metadata, HealType::ECDecode { .. } => HealObjectKind::Decode, + HealType::DeleteMarkerPurge { .. } => HealObjectKind::DeleteMarkerPurge, _ => HealObjectKind::Object, }, bucket: bucket.to_owned(), @@ -634,6 +646,12 @@ impl HealTask { version_id, } => (bucket, object, version_id.as_deref()), HealType::Metadata { bucket, object } => (bucket, object, None), + HealType::DeleteMarkerPurge { + bucket, + object, + version_id, + .. + } => (bucket, object, Some(version_id.as_str())), _ => return None, }; Some(self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index)) @@ -752,6 +770,15 @@ impl HealTask { None => event, } } + HealType::DeleteMarkerPurge { + bucket, + object, + version_id, + .. + } => event + .with_bucket(bucket.as_str()) + .with_object(object.as_str()) + .with_attr("version_id", version_id.as_str()), }; match error { @@ -1091,6 +1118,12 @@ impl HealTask { object, version_id, } => self.heal_ec_decode(bucket, object, version_id.as_deref()).await, + HealType::DeleteMarkerPurge { + bucket, + object, + version_id, + purge, + } => self.heal_delete_marker_purge(bucket, object, version_id, purge).await, HealType::ErasureSet { buckets, set_disk_id } => { self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await } diff --git a/crates/heal/src/heal/task/heal_object.rs b/crates/heal/src/heal/task/heal_object.rs index fe6a4af23..2524f67a0 100644 --- a/crates/heal/src/heal/task/heal_object.rs +++ b/crates/heal/src/heal/task/heal_object.rs @@ -15,6 +15,47 @@ use super::*; impl HealTask { + pub(super) async fn heal_delete_marker_purge( + &self, + bucket: &str, + object: &str, + version_id: &str, + purge: &rustfs_common::mrf_channel::MrfDeleteMarkerPurge, + ) -> Result<()> { + self.check_control_flags().await?; + let heal_opts = HealOpts { + recursive: false, + dry_run: self.options.dry_run, + remove: true, + recreate: false, + scan_mode: HealScanMode::Deep, + update_parity: false, + no_lock: self.options.no_lock, + read_repair: false, + pool: self.options.pool_index, + set: self.options.set_index, + }; + let mut expected = + self.outcome_identity(bucket, object, Some(version_id), self.options.pool_index, self.options.set_index); + expected.bucket_incarnation_id = Some(purge.bucket_incarnation_id); + let storage_result = self + .await_with_control( + self.storage + .purge_delete_marker(bucket, object, version_id, purge, &heal_opts), + ) + .await?; + if let Some(error) = storage_result.error { + return Err(error); + } + if !self.record_verified_storage_receipt(expected, storage_result.receipt).await { + return Err(Error::TaskExecutionFailed { + message: format!("delete-marker purge returned no exact storage proof for {bucket}/{object}/{version_id}"), + }); + } + self.progress.write().await.update_object_progress(1, 1, 0, 0, 0); + Ok(()) + } + // specific heal implementation method #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))] #[hotpath::measure] diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index c5c5e025c..9df0e2dc7 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1783,6 +1783,14 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + require_incarnation_body_digest(&request)?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_delete_version(request, true).await } @@ -4423,7 +4431,7 @@ mod tests { assert_gated!( delete_retired_marker, DeleteVersionRequest { - bucket_incarnation_id: Default::default(), + bucket_incarnation_id: vec![1; 16].into(), disk: disk.clone(), volume: "v".into(), path: "p".into(), @@ -7382,7 +7390,8 @@ mod tests { (vec![0; 16], Some(vec![0; 16]), tonic::Code::InvalidArgument), (vec![1; 15], Some(vec![1; 15]), tonic::Code::InvalidArgument), (vec![2; 16], Some(vec![1; 16]), tonic::Code::PermissionDenied), - (Vec::new(), Some(vec![1; 16]), tonic::Code::PermissionDenied), + (Vec::new(), Some(vec![1; 16]), tonic::Code::InvalidArgument), + (Vec::new(), Some(Vec::new()), tonic::Code::InvalidArgument), (vec![1; 16], Some(vec![1; 16]), tonic::Code::FailedPrecondition), ] { let mut request = Request::new(message.clone()); diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 8095fff4a..c7e73aecc 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -753,6 +753,14 @@ impl NodeService { rustfs_protos::canonical_delete_version_request_body(request.get_ref()), "delete_version", )?; + if require_marker_condition + && Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("retired marker deletion requires a non-nil bucket incarnation")); + } let request = request.into_inner(); let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info,