diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 2edd136de..a8358b4c5 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -387,6 +387,7 @@ pub mod store_list { } pub mod storage { + pub use crate::store::HealWalkVersion; pub use crate::store::{ ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_lock_clients, prewarm_local_disk_id_map, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 71516a7fb..506307977 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -495,6 +495,8 @@ mod read; mod replication; pub(crate) mod shard_source; +pub use ops::heal_walk::HealWalkVersion; + /// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds) /// Defaults to 30 seconds if not set or invalid /// Lock acquisition timeout. Cached: this is consulted on every object diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 97d23d2a9..6eeea3e9a 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -28,6 +28,23 @@ impl SetDisks { object: &str, version_id: &str, opts: &HealOpts, + ) -> disk::error::Result<(HealResultItem, Option)> { + // `allow_meta_regen` is true on the first pass: a version whose data shards + // physically survive (>= data_blocks) but whose xl.meta fell below + // read-quorum is RESCUED (missing xl.meta regenerated) rather than + // dangling-deleted. The re-drive after a rescue sets it false so the + // regeneration can happen at most once (no unbounded recursion). + Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, true)).await + } + + #[allow(clippy::too_many_lines)] + async fn heal_object_with_regen( + &self, + bucket: &str, + object: &str, + version_id: &str, + opts: &HealOpts, + allow_meta_regen: bool, ) -> disk::error::Result<(HealResultItem, Option)> { info!(?opts, "Starting heal_object"); @@ -213,6 +230,22 @@ impl SetDisks { } } + // DATA-SAFETY GUARD (backlog#920, decision 1): before any + // dangling delete, if the version's DATA shards physically + // survive on >= data_blocks disks it is RECONSTRUCTABLE. + // Regenerate the missing xl.meta from a surviving valid + // FileInfo and re-drive the heal instead of destroying a + // recoverable version. Torn writes (< data_blocks data + // shards) fall through to the existing dangling behavior. + if cannot_heal + && allow_meta_regen + && self + .try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks) + .await? + { + return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await; + } + if cannot_heal { let total_disks = parts_metadata.len(); let healthy_count = total_disks.saturating_sub(disks_to_heal_count); @@ -644,6 +677,19 @@ impl SetDisks { } } Err(err) => { + // DATA-SAFETY GUARD (backlog#920, decision 1): meta quorum failed, + // but the version's DATA may still physically survive on enough + // disks (xl.meta lost on > parity disks while part files remain). + // Rescue it by regenerating the missing xl.meta and re-driving heal + // instead of dangling-deleting a reconstructable version. + if allow_meta_regen + && self + .try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks) + .await? + { + return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await; + } + let data_errs_by_part = HashMap::new(); match self .delete_if_dangling( @@ -677,6 +723,131 @@ impl SetDisks { } } + /// backlog#920 (decision 1): rescue a version that meta-quorum logic would + /// otherwise dangling-DELETE, when its DATA is still reconstructable. + /// + /// Returns `Ok(true)` if the version was rescued (missing xl.meta regenerated + /// on at least one disk, so a re-driven heal can reconstruct it), `Ok(false)` + /// to fall through to the existing dangling-delete behavior. + /// + /// Recoverability is computed by physically probing part files across ALL + /// disks in the set with `check_parts` — including disks whose xl.meta is + /// absent (a lost xl.meta does not lose the sibling `part.*` data). If at + /// least `data_blocks` disks hold every part of a surviving valid FileInfo, + /// the object is EC-reconstructable, so we regenerate that FileInfo's xl.meta + /// on every disk whose metadata is absent (via `write_metadata`, which merges + /// into any existing xl.meta). Delete markers, remote/transitioned versions, + /// and genuine torn writes (< `data_blocks` surviving data shards) are NOT + /// rescued — they keep the current dangling-delete-after-grace behavior, so no + /// regression on those paths. + async fn try_regenerate_recoverable_meta( + &self, + bucket: &str, + object: &str, + parts_metadata: &[FileInfo], + errs: &[Option], + disks: &[Option], + ) -> disk::error::Result { + // A surviving valid, non-deleted, non-remote data FileInfo to rebuild from. + let Some(surviving) = parts_metadata + .iter() + .find(|fi| fi.is_valid() && !fi.deleted && !fi.is_remote()) + .cloned() + else { + return Ok(false); + }; + + // Without a data_dir + parts there is no data to prove recoverable. + if surviving.data_dir.is_none() || surviving.parts.is_empty() { + return Ok(false); + } + let data_blocks = surviving.erasure.data_blocks; + if data_blocks == 0 { + return Ok(false); + } + + // Physically probe part presence on EVERY online disk using the surviving + // FileInfo's data_dir/parts. `check_parts` stats `object//part.N` + // directly, so it counts disks that still hold the data even if their + // xl.meta was deleted. + let mut available = 0usize; + for disk in disks.iter().flatten() { + if let Ok(resp) = disk.check_parts(bucket, object, &surviving).await + && !resp.results.is_empty() + && resp.results.iter().all(|r| *r == CHECK_PART_SUCCESS) + { + available += 1; + } + } + + // Torn write: fewer than data_blocks surviving data shards is genuinely + // unrecoverable — preserve the current dangling behavior (no resurrection). + if available < data_blocks { + debug!( + bucket, + object, + available, + data_blocks, + "heal_object: version not reconstructable (torn write), keeping dangling behavior" + ); + return Ok(false); + } + + // Reconstructable: regenerate the surviving xl.meta on every disk whose + // metadata is absent so the version regains read-quorum. Each disk gets its + // OWN shard index: the disk at physical position `index` holds shard + // `distribution[index]` (mirrors `shuffle_disks` + the write path's + // `erasure.index = shuffled_pos + 1`). Copying the surviving disk's index + // verbatim would write an inconsistent xl.meta that the re-heal then treats + // as corrupt. + let distribution = &surviving.erasure.distribution; + let mut wrote = 0usize; + for (index, disk) in disks.iter().enumerate() { + let Some(disk) = disk else { continue }; + let meta_absent = matches!( + errs.get(index).and_then(Option::as_ref), + Some(DiskError::FileNotFound | DiskError::FileVersionNotFound) + ) || !parts_metadata.get(index).map(FileInfo::is_valid).unwrap_or(false); + if !meta_absent { + continue; + } + // Without a known shard index for this position we cannot write a + // consistent xl.meta; leave it for the normal heal to reconstruct. + let Some(&shard_index) = distribution.get(index) else { + continue; + }; + let mut regen = surviving.clone(); + regen.fresh = false; // merge into any existing xl.meta on the disk + regen.erasure.index = shard_index; + match disk.write_metadata("", bucket, object, regen).await { + Ok(()) => wrote += 1, + Err(e) => { + warn!( + bucket, + object, + disk_index = index, + error = %e, + "heal_object: failed to regenerate recoverable xl.meta on disk" + ); + } + } + } + + if wrote == 0 { + return Ok(false); + } + + info!( + bucket, + object, + available, + data_blocks, + regenerated_meta_disks = wrote, + "heal_object: rescued reconstructable sub-quorum version by regenerating xl.meta" + ); + Ok(true) + } + pub(in crate::set_disk) async fn heal_object_dir_locked( &self, bucket: &str, diff --git a/crates/ecstore/src/set_disk/ops/heal_walk.rs b/crates/ecstore/src/set_disk/ops/heal_walk.rs new file mode 100644 index 000000000..d3391b5be --- /dev/null +++ b/crates/ecstore/src/set_disk/ops/heal_walk.rs @@ -0,0 +1,306 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Per-erasure-set disk-walk UNION enumerator for heal (backlog#920). +//! +//! B5's heal enumeration lists versions through the READ-QUORUM metadata view +//! (`list_object_versions`), so a version present on FEWER than read-quorum disks +//! is never enumerated and therefore never healed. This module walks each disk in +//! the set directly (mirroring MinIO cmd/global-heal.go `healErasureSet`: +//! `listPathRaw` with `objQuorum = 1` feeding `mergeXLV2Versions`) and surfaces +//! every `(object, version)` present on ANY disk, feeding each to the existing +//! per-version `SetDisks::heal_object`. + +use super::super::*; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +/// A single `(object, version)` unit surfaced by the disk-walk union enumerator. +/// +/// `is_delete_marker` is OBSERVABILITY-ONLY (metrics / logging / e2e assertions); +/// it must not gate healing logic — the delete-marker vs data path is chosen +/// inside `ops/heal.rs` from the resolved latest metadata. `version_id` is +/// normalized (nil/absent UUID => `None`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HealWalkVersion { + /// object key + pub name: String, + /// normalized version id (`None` when the version is nil/absent) + pub version_id: Option, + /// whether this version is a delete marker (observability only) + pub is_delete_marker: bool, +} + +/// One collected object (a single sorted key) with all its expanded versions. +#[derive(Debug, Clone)] +struct HealWalkObject { + name: String, + versions: Vec, +} + +/// Shared collector fed by the `agreed`/`partial` callbacks of `list_path_raw`. +/// `list_path_raw` emits at most one callback per distinct object key in sorted +/// order, so each successful ingest corresponds to exactly one new object. +struct HealWalkCollector { + bucket: String, + batch_objects: usize, + version_budget: usize, + objects: Mutex>, + version_total: AtomicUsize, + truncated: AtomicBool, + cancel: CancellationToken, +} + +impl HealWalkCollector { + /// Expand one resolved entry into its versions and record it. Cancels the + /// walk once EITHER page bound (distinct object names OR expanded versions) + /// is met — always at a sorted object-key boundary so a heavily-versioned + /// object is never split across pages. + fn ingest(&self, entry: MetaCacheEntry) { + // Skip pure directory entries; they carry no versions to heal here. + if entry.is_dir() { + return; + } + + // Expand to one HealWalkVersion per FileInfo. KEEP remote/transitioned and + // free-version records: their LOCAL xl.meta may still need healing. + let fiv = match entry.file_info_versions_with_free_versions(&self.bucket) { + Ok(fiv) => fiv, + Err(err) => { + debug!(entry = %entry.name, error = ?err, "heal disk-walk skipped entry with unreadable versions"); + return; + } + }; + + let mut versions = Vec::with_capacity(fiv.versions.len() + fiv.free_versions.len()); + for fi in fiv.versions.iter().chain(fiv.free_versions.iter()) { + versions.push(HealWalkVersion { + name: entry.name.clone(), + // Normalize: nil/absent version id => None. + version_id: fi.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()), + is_delete_marker: fi.deleted, + }); + } + + if versions.is_empty() { + return; + } + + let added = versions.len(); + let (objs_len, ver_total) = { + let mut objects = self.objects.lock().unwrap(); + objects.push(HealWalkObject { + name: entry.name, + versions, + }); + let objs_len = objects.len(); + let ver_total = self.version_total.fetch_add(added, Ordering::SeqCst) + added; + (objs_len, ver_total) + }; + + // Bound at an object boundary (this object is fully included). + if objs_len >= self.batch_objects || ver_total >= self.version_budget { + self.truncated.store(true, Ordering::SeqCst); + self.cancel.cancel(); + } + } +} + +/// Finalize a collected disk-walk page: compute the resume cursor and apply +/// inclusive-forward de-overlap. +/// +/// `next_forward` is derived from the PRE-de-overlap last collected object name +/// (the walk's `forward_to` is inclusive, so the next page re-reads that exact +/// key and drops it here). Leading objects whose name == `forward_to` are dropped +/// to avoid re-healing the boundary object. +fn finalize_heal_walk_page( + mut objects: Vec, + forward_to: Option<&str>, + truncated: bool, +) -> (Vec, Option, bool) { + let next_forward = objects.last().map(|o| o.name.clone()); + + if let Some(fw) = forward_to { + while objects.first().map(|o| o.name.as_str()) == Some(fw) { + objects.remove(0); + } + } + + let versions: Vec = objects.into_iter().flat_map(|o| o.versions).collect(); + + if truncated { + (versions, next_forward, true) + } else { + (versions, None, false) + } +} + +impl SetDisks { + /// Walk every disk in this set and return one page of the cross-disk UNION of + /// versions (see module docs). Returns `(versions, next_forward, truncated)`. + /// + /// - `batch_objects` (>= 2) and `version_budget` are a DUAL page bound: the + /// walk stops at the first sorted object-key boundary where EITHER the + /// distinct-object-name count reaches `batch_objects` or the expanded + /// version count reaches `version_budget`. + /// - `forward_to` resumes the walk (inclusive); the boundary object is + /// de-overlapped here so no version is healed twice across pages. + /// + /// `min_disks: 1` means the page is produced from WHATEVER disks respond, so a + /// version on a single surviving disk is still surfaced. + pub(crate) async fn heal_walk_versions_page( + &self, + bucket: &str, + prefix: &str, + forward_to: Option<&str>, + batch_objects: usize, + version_budget: usize, + ) -> disk::error::Result<(Vec, Option, bool)> { + assert!(batch_objects >= 2, "heal_walk_versions_page requires batch_objects >= 2"); + + let disks = self.get_disks_internal().await; + + let collector = Arc::new(HealWalkCollector { + bucket: bucket.to_string(), + batch_objects, + version_budget: version_budget.max(1), + objects: Mutex::new(Vec::new()), + version_total: AtomicUsize::new(0), + truncated: AtomicBool::new(false), + cancel: CancellationToken::new(), + }); + + let agreed_collector = collector.clone(); + let partial_collector = collector.clone(); + let partial_bucket = bucket.to_string(); + + let filter_prefix = if prefix.is_empty() { None } else { Some(prefix.to_string()) }; + + let opts = ListPathRawOptions { + disks, + bucket: bucket.to_string(), + path: String::new(), + recursive: true, + incl_deleted: true, + filter_prefix, + forward_to: forward_to.map(str::to_string), + min_disks: 1, + report_not_found: false, + per_disk_limit: 0, + skip_walkdir_total_timeout: true, + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let collector = agreed_collector.clone(); + Box::pin(async move { + collector.ingest(entry); + }) + })), + partial: Some(Box::new(move |entries: MetaCacheEntries, _errs: &[Option]| { + let collector = partial_collector.clone(); + let bucket = partial_bucket.clone(); + Box::pin(async move { + // objQuorum = 1: take the cross-disk union of every version on + // any disk. Fall back to the first present entry if the merge + // somehow yields nothing. + let entry = entries.resolve_union(&bucket).or_else(|| entries.first_found().0); + if let Some(entry) = entry { + collector.ingest(entry); + } + }) + })), + finished: None, + ..Default::default() + }; + + // Drive the walk. A tolerated missing-path / not-found is treated as an + // empty page rather than an error (nothing to heal on this prefix). + match list_path_raw(collector.cancel.clone(), opts).await { + Ok(()) => {} + Err(DiskError::FileNotFound) | Err(DiskError::VolumeNotFound) => { + debug!(bucket, prefix, "heal disk-walk treated missing path as empty page"); + } + Err(err) => return Err(err), + } + + let objects = std::mem::take(&mut *collector.objects.lock().unwrap()); + let truncated = collector.truncated.load(Ordering::SeqCst); + Ok(finalize_heal_walk_page(objects, forward_to, truncated)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn version(name: &str, id: &str, dm: bool) -> HealWalkVersion { + HealWalkVersion { + name: name.to_string(), + version_id: Some(id.to_string()), + is_delete_marker: dm, + } + } + + fn object_with_versions(name: &str, count: usize) -> HealWalkObject { + HealWalkObject { + name: name.to_string(), + versions: (0..count).map(|i| version(name, &format!("{name}-v{i}"), false)).collect(), + } + } + + /// A single heavily-versioned object whose version count exceeds the budget + /// is returned WHOLE in one page (never split), and the page is flagged + /// truncated with a resume cursor pointing at that object. + #[test] + fn union_page_bounded_by_version_budget_on_heavily_versioned_object() { + let big = object_with_versions("big.bin", 50); + // The collector cancels AFTER completing the object, so the buffer holds + // exactly this one object even though 50 >> the version budget of 10. + let (versions, next_forward, truncated) = finalize_heal_walk_page(vec![big], None, true); + + assert_eq!(versions.len(), 50, "the heavily-versioned object must not be split across pages"); + assert!(truncated, "exceeding the version budget must mark the page truncated"); + assert_eq!( + next_forward.as_deref(), + Some("big.bin"), + "resume cursor must point at the boundary object" + ); + } + + /// Inclusive-forward de-overlap: the boundary object re-read on the next page + /// (name == forward_to) is dropped, so its versions are not healed twice. + #[test] + fn union_page_de_overlaps_inclusive_forward_boundary() { + let a = object_with_versions("a.bin", 2); + let b = object_with_versions("b.bin", 3); + // forward_to == "a.bin": the walk re-reads a.bin (inclusive) then b.bin. + let (versions, next_forward, truncated) = finalize_heal_walk_page(vec![a, b], Some("a.bin"), true); + + assert!( + versions.iter().all(|v| v.name == "b.bin"), + "the boundary object a.bin must be de-overlapped, got {versions:?}" + ); + assert_eq!(versions.len(), 3); + assert_eq!(next_forward.as_deref(), Some("b.bin")); + assert!(truncated); + } + + /// A non-truncated final page clears the resume cursor. + #[test] + fn union_page_non_truncated_clears_cursor() { + let a = object_with_versions("a.bin", 1); + let (versions, next_forward, truncated) = finalize_heal_walk_page(vec![a], None, false); + assert_eq!(versions.len(), 1); + assert_eq!(next_forward, None, "a complete final page must not carry a resume cursor"); + assert!(!truncated); + } +} diff --git a/crates/ecstore/src/set_disk/ops/mod.rs b/crates/ecstore/src/set_disk/ops/mod.rs index 9b45188ef..44ea75cb9 100644 --- a/crates/ecstore/src/set_disk/ops/mod.rs +++ b/crates/ecstore/src/set_disk/ops/mod.rs @@ -19,6 +19,7 @@ pub(crate) mod bucket; pub(crate) mod heal; +pub(crate) mod heal_walk; pub(crate) mod list; pub(crate) mod locking; pub(crate) mod multipart; diff --git a/crates/ecstore/src/store/heal_walk.rs b/crates/ecstore/src/store/heal_walk.rs new file mode 100644 index 000000000..39d50007d --- /dev/null +++ b/crates/ecstore/src/store/heal_walk.rs @@ -0,0 +1,50 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Store-level entry point for the per-erasure-set disk-walk UNION heal +//! enumerator (backlog#920). Bounds-checks `(pool_idx, set_idx)` (mirroring +//! `handle_get_disks`) and delegates to the target `SetDisks`. + +use super::*; +pub use crate::set_disk::HealWalkVersion; + +impl ECStore { + /// Walk a specific erasure set's disks and return one page of the cross-disk + /// UNION of versions (see `SetDisks::heal_walk_versions_page`). `pool_idx` and + /// `set_idx` are bounds-checked against the live pool topology. + #[instrument(level = "debug", skip(self))] + #[allow(clippy::too_many_arguments)] + pub async fn heal_walk_versions_page( + &self, + pool_idx: usize, + set_idx: usize, + bucket: &str, + prefix: &str, + forward_to: Option<&str>, + batch_objects: usize, + version_budget: usize, + ) -> Result<(Vec, Option, bool)> { + if pool_idx >= self.pools.len() || set_idx >= self.pools[pool_idx].disk_set.len() { + return Err(Error::other(format!( + "heal disk-walk: invalid erasure set (pool index {pool_idx}, set index {set_idx}, pool count {})", + self.pools.len() + ))); + } + + self.pools[pool_idx].disk_set[set_idx] + .heal_walk_versions_page(bucket, prefix, forward_to, batch_objects, version_budget) + .await + .map_err(Error::from) + } +} diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index d75acef3f..cab791ce0 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -150,6 +150,8 @@ const MAX_UPLOADS_LIST: usize = 10000; mod bucket; mod heal; +mod heal_walk; +pub use heal_walk::HealWalkVersion; mod init; pub(crate) mod init_format; mod list; diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index 0279cf92d..b5525a0bf 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -4018,6 +4018,69 @@ mod tests { assert_eq!(merged, vec![valid]); } + /// Build a single distinct Object version header for the union tests. + fn union_version(version_u128: u128, mod_unix: i64) -> FileMetaShallowVersion { + FileMetaShallowVersion { + header: FileMetaVersionHeader { + version_id: Some(Uuid::from_u128(version_u128)), + mod_time: Some(OffsetDateTime::from_unix_timestamp(mod_unix).expect("valid timestamp")), + signature: [0x11, 0x22, 0x33, 0x44], + version_type: VersionType::Object, + flags: 0, + ec_n: 4, + ec_m: 2, + }, + meta: Vec::new(), + } + } + + /// backlog#920: quorum==1 (union) must surface EVERY version present on ANY + /// per-disk stream, even when the per-disk version sets are fully DISJOINT + /// (a version living on a single disk). This is the enumeration guarantee the + /// sub-quorum disk-walk depends on. + #[test] + fn merge_union_quorum1_yields_full_union_over_disjoint_disks() { + let a = union_version(0xA, 1_705_312_300); + let b = union_version(0xB, 1_705_312_200); + let c = union_version(0xC, 1_705_312_100); + + // Three disks, each holding exactly ONE distinct version (disjoint sets). + let disks = [vec![a], vec![b], vec![c]]; + let merged = merge_file_meta_versions(1, true, 0, &disks); + + let ids: std::collections::HashSet> = merged.iter().map(|v| v.header.version_id).collect(); + assert_eq!(merged.len(), 3, "union must retain all three disjoint versions: {merged:?}"); + assert!(ids.contains(&Some(Uuid::from_u128(0xA)))); + assert!(ids.contains(&Some(Uuid::from_u128(0xB)))); + assert!(ids.contains(&Some(Uuid::from_u128(0xC)))); + + // Contrast: read-quorum (2) over the same disjoint streams surfaces NONE, + // because no version reaches two agreeing disks. This is the exact gap. + let read_quorum = merge_file_meta_versions(2, true, 0, &disks); + assert!( + read_quorum.is_empty(), + "read-quorum merge must drop every sub-quorum version, got {read_quorum:?}" + ); + } + + /// The equal-modTime / distinct-versionId retain-and-advance branch: two + /// versions sharing a mod_time but with different ids must BOTH survive the + /// union merge (they are not collapsed as duplicates). + #[test] + fn merge_union_quorum1_retains_equal_modtime_distinct_version_ids() { + let same_time = 1_705_312_300; + let a = union_version(0xAA, same_time); + let b = union_version(0xBB, same_time); + + let disks = [vec![a], vec![b]]; + let merged = merge_file_meta_versions(1, true, 0, &disks); + + let ids: std::collections::HashSet> = merged.iter().map(|v| v.header.version_id).collect(); + assert_eq!(merged.len(), 2, "equal-modTime distinct-id versions must both survive: {merged:?}"); + assert!(ids.contains(&Some(Uuid::from_u128(0xAA)))); + assert!(ids.contains(&Some(Uuid::from_u128(0xBB)))); + } + #[test] fn meta_object_init_free_version_rejects_invalid_tier_free_version_id() { let mut sys = HashMap::new(); diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 7b1c971ee..3a8d18e38 100644 --- a/crates/filemeta/src/metacache.rs +++ b/crates/filemeta/src/metacache.rs @@ -343,6 +343,32 @@ impl MetaCacheEntries { self.resolve_inner(params, true) } + /// Resolve the cross-disk UNION of every version present on ANY disk slot. + /// + /// This mirrors MinIO's global heal enumeration (cmd/global-heal.go + /// `healErasureSet` -> `listPathRaw` with `objQuorum = 1` feeding + /// `mergeEntries`/`mergeXLV2Versions`): at quorum 1 the merge is forced + /// strict and returns the union of all per-disk version streams, so a version + /// that survives on FEWER than read-quorum disks is still surfaced for + /// healing. Read-quorum resolution (`resolve` with `obj_quorum >= 2`) would + /// silently drop such a sub-quorum version, which is exactly the durability + /// gap this enumerator closes (backlog#920). + /// + /// `bucket` is only used to tag the merged entry; `strict:false` lets the + /// non-strict header reconciliation collapse equal-but-differently-signed + /// replicas of the SAME version id while still retaining genuinely distinct + /// version ids. + pub fn resolve_union(&self, bucket: &str) -> Option { + self.resolve(MetadataResolutionParams { + dir_quorum: 1, + obj_quorum: 1, + requested_versions: 0, + bucket: bucket.to_string(), + strict: false, + candidates: Vec::new(), + }) + } + fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option { if self.0.is_empty() { debug!( @@ -1480,6 +1506,92 @@ mod tests { } } + /// Build an entry holding a single object version with an explicit version id + /// and mod_time, so a set of these can model DISJOINT per-disk version sets. + fn metacache_entry_single_version(version_u128: u128, mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry { + let mut metadata = HashMap::new(); + metadata.insert("etag".to_string(), etag.to_string()); + + let mut fi = FileInfo::new("object", 4, 2); + fi.volume = "bucket".to_string(); + fi.name = "object".to_string(); + fi.version_id = Some(Uuid::from_u128(version_u128)); + fi.versioned = true; + fi.size = 1; + fi.mod_time = Some(mod_time); + fi.metadata = metadata; + + let mut meta = FileMeta::new(); + meta.add_version(fi).expect("test file metadata should accept object version"); + let encoded = meta.marshal_msg().expect("test file metadata should marshal"); + + MetaCacheEntry { + name: "object".to_string(), + metadata: encoded, + cached: Some(meta), + reusable: false, + } + } + + /// backlog#920: `resolve_union` surfaces a version present on a SINGLE slot + /// among four, while `resolve` at read-quorum (obj_quorum=2) drops it. This + /// documents the exact sub-quorum durability gap the disk-walk closes. + #[test] + fn resolve_union_surfaces_single_disk_version() { + let t0 = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let t1 = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + + // Slot 0 carries a UNIQUE version (0xBEEF) present nowhere else; the other + // three slots agree on a shared version (0xCAFE). Read-quorum sees only + // the shared one; union must see BOTH. + let unique = metacache_entry_single_version(0xBEEF, t1, "unique-etag"); + let shared_a = metacache_entry_single_version(0xCAFE, t0, "shared-etag"); + let shared_b = metacache_entry_single_version(0xCAFE, t0, "shared-etag"); + let shared_c = metacache_entry_single_version(0xCAFE, t0, "shared-etag"); + + let entries = || { + MetaCacheEntries(vec![ + Some(unique.clone()), + Some(shared_a.clone()), + Some(shared_b.clone()), + Some(shared_c.clone()), + ]) + }; + + let union = entries().resolve_union("bucket").expect("union must resolve an entry"); + let union_meta = union.cached.expect("union entry keeps cached metadata"); + let union_ids: std::collections::HashSet> = + union_meta.versions.iter().map(|v| v.header.version_id).collect(); + assert!( + union_ids.contains(&Some(Uuid::from_u128(0xBEEF))), + "union must surface the single-slot version: {union_ids:?}" + ); + assert!( + union_ids.contains(&Some(Uuid::from_u128(0xCAFE))), + "union must also keep the shared version: {union_ids:?}" + ); + + // Read-quorum resolution (obj_quorum=2) drops the single-slot version. + let read_quorum = entries() + .resolve(MetadataResolutionParams { + obj_quorum: 2, + dir_quorum: 2, + strict: false, + ..Default::default() + }) + .expect("read-quorum must still resolve the shared version"); + let rq_meta = read_quorum.cached.expect("read-quorum entry keeps cached metadata"); + let rq_ids: std::collections::HashSet> = rq_meta.versions.iter().map(|v| v.header.version_id).collect(); + assert!( + !rq_ids.contains(&Some(Uuid::from_u128(0xBEEF))), + "read-quorum must DROP the single-slot version (the gap): {rq_ids:?}" + ); + assert!( + rq_ids.contains(&Some(Uuid::from_u128(0xCAFE))), + "read-quorum must keep the shared version: {rq_ids:?}" + ); + } + #[test] fn resolve_rejects_partial_latest_and_returns_committed_previous_metadata() { let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 66d5b9f67..2f4c6f87d 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -579,12 +579,26 @@ impl ErasureSetHealer { Self::effective_heal_page_object_concurrency_for_source(self.source, self.heal_opts.scan_mode); let in_flight = Arc::new(AtomicUsize::new(0)); + // backlog#920: select the per-erasure-set DISK-WALK union enumerator when + // the scan is Deep OR the request came from AutoHeal — these are the paths + // that must repair sub-quorum-but-reconstructable versions. Every other + // (Normal, non-AutoHeal) request keeps the unchanged B5 read-quorum path, + // which stays the default. + let use_disk_walk = + matches!(self.heal_opts.scan_mode, HealScanMode::Deep) || matches!(self.source, HealRequestSource::AutoHeal); + loop { // Get one page of object versions - let (objects, next_token, is_truncated) = self - .storage - .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()) - .await?; + let (objects, next_token, is_truncated) = if use_disk_walk { + self.storage + .list_versions_for_heal_page_disk_walk(set_disk_id, bucket, "", continuation_token.as_deref()) + .await? + } else { + self.storage + .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()) + .await? + }; + let page_is_empty = objects.is_empty(); let checkpoint = checkpoint_manager.get_checkpoint().await; let page_resume_index = *current_object_index; let semaphore = Arc::new(Semaphore::new(page_concurrency_limit)); @@ -779,6 +793,15 @@ impl ErasureSetHealer { break; } + // Anti-loop guard: an empty page reported as truncated cannot advance + // the cursor (there is no last identity to move past), so treat it as a + // non-advancing listing and abort rather than spin forever. + if page_is_empty { + return Err(Error::other(format!( + "Erasure set heal listing for bucket {bucket} returned an empty page marked truncated; aborting to avoid an infinite loop" + ))); + } + // Anti-loop guard: if the backend keeps reporting truncation but the // last version identity did not advance, we would spin forever. if page_last.is_some() && page_last == previous_page_last { diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index ac93fe778..a9d9f04ca 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -182,6 +182,78 @@ pub(crate) fn decode_heal_token(token: &str) -> (Option, Option) (payload.m, payload.v) } +const DISK_WALK_TOKEN_PREFIX: &str = "dw1:"; + +/// Encode the disk-walk resume cursor (`next_forward` object key) into an opaque +/// continuation token: `"dw1:" + base64url_nopad(forward)`. +/// +/// The `dw1:` namespace is DISJOINT from the B5 `v1:` token namespace so the two +/// enumerators can never misread each other's cursor: a `dw1:` token decodes to +/// `(None, None)` under the B5 decoder, and a `v1:` token decodes to `None` here. +pub(crate) fn encode_disk_walk_token(next_forward: &str) -> String { + format!("{DISK_WALK_TOKEN_PREFIX}{}", URL_SAFE_NO_PAD.encode(next_forward.as_bytes())) +} + +/// Decode a disk-walk continuation token back into the `next_forward` object key. +/// +/// TOTAL function: an empty token, a missing `"dw1:"` prefix (including a foreign +/// B5 `v1:` token), invalid base64, or invalid UTF-8 all decode to `None` (start +/// the walk from the beginning). This makes a restart across an enumerator switch +/// idempotent rather than corrupting. +pub(crate) fn decode_disk_walk_token(token: &str) -> Option { + if token.is_empty() { + return None; + } + + let Some(encoded) = token.strip_prefix(DISK_WALK_TOKEN_PREFIX) else { + warn!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "decode_disk_walk_token", + state = "foreign_or_missing_prefix", + "Disk-walk continuation token missing dw1 prefix; restarting walk" + ); + return None; + }; + + let bytes = match URL_SAFE_NO_PAD.decode(encoded) { + Ok(bytes) => bytes, + Err(e) => { + warn!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "decode_disk_walk_token", + state = "bad_base64", + error = %e, + "Disk-walk continuation token has invalid base64; restarting walk" + ); + return None; + } + }; + + match String::from_utf8(bytes) { + Ok(forward) if !forward.is_empty() => Some(forward), + Ok(_) => None, + Err(e) => { + warn!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "decode_disk_walk_token", + state = "bad_utf8", + error = %e, + "Disk-walk continuation token has invalid utf8; restarting walk" + ); + None + } + } +} + /// A single object version to heal. /// /// `is_delete_marker` is OBSERVABILITY-ONLY (metrics / logging / e2e @@ -298,6 +370,26 @@ pub trait HealStorageAPI: Send + Sync { continuation_token: Option<&str>, ) -> Result<(Vec, Option, bool)>; + /// List versions for healing via a per-erasure-set DISK-WALK union enumerator + /// (backlog#920). Unlike `list_objects_for_heal_page` (which reflects only the + /// READ-QUORUM metadata view via `list_object_versions`), this surfaces every + /// `(object, version)` present on ANY disk in the set identified by + /// `set_disk_id`, so sub-quorum-but-reconstructable versions are healed. + /// + /// The continuation token uses the disjoint `"dw1:"` namespace. The DEFAULT + /// implementation falls back to the read-quorum listing so mock/alternate + /// storages keep compiling and behaving; `ECStoreHealStorage` overrides it + /// with the real disk walk. + async fn list_versions_for_heal_page_disk_walk( + &self, + _set_disk_id: &str, + bucket: &str, + prefix: &str, + continuation_token: Option<&str>, + ) -> Result<(Vec, Option, bool)> { + self.list_objects_for_heal_page(bucket, prefix, continuation_token).await + } + /// Get disk for resume functionality async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result; } @@ -1363,6 +1455,93 @@ impl HealStorageAPI for ECStoreHealStorage { Ok((page_objects, next_token, list_info.is_truncated)) } + async fn list_versions_for_heal_page_disk_walk( + &self, + set_disk_id: &str, + bucket: &str, + prefix: &str, + continuation_token: Option<&str>, + ) -> Result<(Vec, Option, bool)> { + // Per-page bounds for the disk-walk union enumerator. Objects are atomic + // (never split across pages), so version_budget only bounds how many + // versions accumulate before the page is cut at the next object boundary. + const BATCH_OBJECTS: usize = 1000; + const VERSION_BUDGET: usize = 10_000; + + debug!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "list_versions_for_heal_page_disk_walk", + set_disk_id, + bucket, + prefix, + continuation_token = ?continuation_token, + state = "started", + "Heal storage disk-walk union enumeration started" + ); + + let (pool_idx, set_idx) = crate::heal::utils::parse_set_disk_id(set_disk_id)?; + // Decode the dw1: cursor into the forward_to object key. Malformed/foreign + // tokens restart the walk from the beginning (decode_disk_walk_token is total). + let forward_to = decode_disk_walk_token(continuation_token.unwrap_or("")); + + let (versions, next_forward, is_truncated) = self + .ecstore + .heal_walk_versions_page(pool_idx, set_idx, bucket, prefix, forward_to.as_deref(), BATCH_OBJECTS, VERSION_BUDGET) + .await + .map_err(|e| { + error!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "list_versions_for_heal_page_disk_walk", + set_disk_id, + bucket, + prefix, + result = "failed", + error = %e, + "Heal storage disk-walk union enumeration failed" + ); + Error::other(e) + })?; + + let page_objects: Vec = versions + .into_iter() + .map(|v| HealListItem { + name: v.name, + version_id: v.version_id, + is_delete_marker: v.is_delete_marker, + }) + .collect(); + let page_count = page_objects.len(); + + let next_token = if is_truncated { + next_forward.map(|fw| encode_disk_walk_token(&fw)) + } else { + None + }; + + debug!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "list_versions_for_heal_page_disk_walk", + set_disk_id, + bucket, + prefix, + version_count = page_count, + is_truncated, + state = "page_loaded", + "Heal storage disk-walk union enumeration page loaded" + ); + + Ok((page_objects, next_token, is_truncated)) + } + async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result { debug!( target: "rustfs::heal::storage", @@ -1411,8 +1590,8 @@ impl HealStorageAPI for ECStoreHealStorage { mod tests { use super::super::StorageError; use super::{ - decode_heal_token, encode_heal_token, is_transient_object_exists_error, is_transient_object_exists_message, - next_heal_listing_token, + decode_disk_walk_token, decode_heal_token, encode_disk_walk_token, encode_heal_token, is_transient_object_exists_error, + is_transient_object_exists_message, next_heal_listing_token, }; use base64::Engine as _; @@ -1481,6 +1660,37 @@ mod tests { assert_eq!(decode_heal_token(&token), (None, None), "version-only marker must coerce to (None, None)"); } + #[test] + fn disk_walk_cursor_round_trip_and_foreign_token_restarts() { + // Round-trip: a real object key survives encode -> decode. + let token = encode_disk_walk_token("some/deep/object.bin"); + assert!(token.starts_with("dw1:")); + assert_eq!(decode_disk_walk_token(&token), Some("some/deep/object.bin".to_string())); + + // Empty / garbage / missing-prefix all restart the walk (None). + assert_eq!(decode_disk_walk_token(""), None); + assert_eq!(decode_disk_walk_token("dw1:!!!not-base64!!!"), None); + assert_eq!(decode_disk_walk_token("no-prefix-here"), None); + + // CROSS-DECODER ISOLATION (both directions): a dw1 token must not be + // misread as a B5 (marker, version_marker) pair, and a v1 token must not + // be misread as a disk-walk forward cursor. + let dw = encode_disk_walk_token("obj/key"); + assert_eq!( + decode_heal_token(&dw), + (None, None), + "a dw1: token must decode to (None, None) under the B5 decoder" + ); + + let v1 = encode_heal_token(Some("obj/key"), Some("v-123")); + assert!(v1.starts_with("v1:")); + assert_eq!( + decode_disk_walk_token(&v1), + None, + "a v1: token must decode to None under the disk-walk decoder" + ); + } + #[test] fn transient_object_exists_message_matches_lock_quorum_failures() { assert!(is_transient_object_exists_message( diff --git a/crates/heal/tests/heal_b920_subquorum_union_test.rs b/crates/heal/tests/heal_b920_subquorum_union_test.rs new file mode 100644 index 000000000..d18e78033 --- /dev/null +++ b/crates/heal/tests/heal_b920_subquorum_union_test.rs @@ -0,0 +1,568 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! backlog#920: real-disk e2e suite proving the per-erasure-set DISK-WALK UNION +//! heal enumerator surfaces (and heals) sub-quorum versions that the B5 +//! read-quorum enumeration (`list_object_versions`) omits, AND that the +//! data-safety guard (decision 1) regenerates a lost xl.meta for a +//! reconstructable version instead of dangling-DELETING it. +//! +//! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks. +//! Every test is `#[serial]`; under `cargo nextest` each runs in its own process. + +use http::HeaderMap; +use rustfs_common::heal_channel::{HealOpts, HealScanMode}; +use rustfs_heal::heal::storage::{ + ECStoreHealStorage, HealListItem, HealObjectOptions as ObjectOptions, HealPutObjReader as PutObjReader, HealStorageAPI, +}; +use serial_test::serial; +use std::{ + path::{Path, PathBuf}, + sync::{Arc, Once}, +}; +use tokio::fs; +use tokio_util::sync::CancellationToken; +use walkdir::WalkDir; + +mod storage_api; + +use storage_api::integration::{ + BucketOperations, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _, + PoolEndpoints, init_bucket_metadata_sys, init_local_disks, +}; + +/// 256 KiB + change: large enough to be stored as non-inline erasure shards, so +/// deleting the `xl.meta` file does NOT delete the data (the `part.*` shards live +/// in a sibling data-dir). This is required to exercise the "meta lost, data +/// present" rescue path — an inline object would lose its data with its xl.meta. +const NON_INLINE_TEST_DATA_SIZE: usize = 256 * 1024 + 137; + +const SET_DISK_ID: &str = "pool_0_set_0"; +const GRACE_ENV: &str = "RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS"; + +fn versioned_test_data(seed: u8) -> Vec { + (0..NON_INLINE_TEST_DATA_SIZE) + .map(|idx| ((idx + seed as usize) % 251) as u8) + .collect() +} + +static INIT: Once = Once::new(); + +fn init_tracing() { + INIT.call_once(|| { + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .try_init(); + }); +} + +/// Disable the dangling-delete grace window so the destructive path is genuinely +/// LIVE in these tests: without the decision-1 guard, a recoverable version WOULD +/// be dangling-deleted here. With grace at its 1h default the delete path would be +/// masked and the test would prove nothing. +fn disable_dangling_grace() { + // Safe under nextest: each test runs in its own process and is `#[serial]`. + unsafe { + std::env::set_var(GRACE_ENV, "0"); + } +} + +/// Build a real N-disk single-set `ECStore` + `ECStoreHealStorage`. +async fn setup_test_env_n(n_disks: usize) -> (Vec, Arc, Arc) { + init_tracing(); + + let test_base_dir = format!("/tmp/rustfs_heal_b920_test_{}", uuid::Uuid::new_v4()); + let temp_dir = PathBuf::from(&test_base_dir); + if temp_dir.exists() { + fs::remove_dir_all(&temp_dir).await.ok(); + } + fs::create_dir_all(&temp_dir).await.unwrap(); + + let disk_paths: Vec = (0..n_disks).map(|i| temp_dir.join(format!("disk{}", i + 1))).collect(); + for disk_path in &disk_paths { + fs::create_dir_all(disk_path).await.unwrap(); + } + + let mut endpoints = Vec::new(); + for (i, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + endpoints.push(endpoint); + } + + let pool_endpoints = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: n_disks, + endpoints: Endpoints::from(endpoints), + cmd_line: "test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }; + let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]); + + init_local_disks(endpoint_pools.clone()).await.unwrap(); + + let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) + .await + .unwrap(); + + let buckets_list = ecstore + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .unwrap(); + let buckets = buckets_list.into_iter().map(|v| v.name).collect(); + init_bucket_metadata_sys(ecstore.clone(), buckets).await; + + let heal_storage = Arc::new(ECStoreHealStorage::new(ecstore.clone())); + (disk_paths, ecstore, heal_storage) +} + +async fn create_versioned_bucket(ecstore: &Arc, bucket: &str) { + (**ecstore) + .make_bucket( + bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("failed to create versioned bucket"); +} + +async fn put_versioned(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) -> String { + let mut reader = PutObjReader::from_vec(data.to_vec()); + let opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + let info = (**ecstore) + .put_object(bucket, object, &mut reader, &opts) + .await + .expect("versioned put_object failed"); + info.version_id + .map(|u| u.to_string()) + .expect("versioned put must return a version id") +} + +fn object_dir(disk: &Path, bucket: &str, object: &str) -> PathBuf { + disk.join(bucket).join(object) +} + +/// Count `part.*` data-shard files two levels below the object dir. +fn count_part_files(obj_dir: &Path) -> usize { + if !obj_dir.exists() { + return 0; + } + WalkDir::new(obj_dir) + .min_depth(2) + .max_depth(2) + .into_iter() + .filter_map(Result::ok) + .filter(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false)) + .count() +} + +fn xl_meta_path(obj_dir: &Path) -> PathBuf { + obj_dir.join("xl.meta") +} + +/// Delete ONLY the `xl.meta` file for an object on one disk, leaving its `part.*` +/// data shards intact (models a lost-metadata-but-present-data disk). +fn remove_xl_meta_only(disk: &Path, bucket: &str, object: &str) { + let meta = xl_meta_path(&object_dir(disk, bucket, object)); + assert!(meta.exists(), "xl.meta must exist before removal: {meta:?}"); + std::fs::remove_file(&meta).expect("failed to remove xl.meta"); + // Data dir + part files remain. + assert!( + count_part_files(&object_dir(disk, bucket, object)) >= 1, + "data shards must remain after removing only xl.meta" + ); +} + +fn deep_heal_opts() -> HealOpts { + HealOpts { + recreate: true, + remove: false, + scan_mode: HealScanMode::Deep, + ..Default::default() + } +} + +async fn read_version(ecstore: &Arc, bucket: &str, object: &str, version_id: &str) -> Vec { + let opts = ObjectOptions { + version_id: Some(version_id.to_string()), + ..Default::default() + }; + let mut reader = ecstore + .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) + .await + .expect("failed to open version reader"); + let mut buf = Vec::new(); + tokio::io::copy(&mut reader, &mut buf) + .await + .expect("failed to read version data"); + buf +} + +/// Enumerate every version via the B5 read-quorum listing (walks all pages). +async fn enumerate_b5(heal_storage: &Arc, bucket: &str) -> Vec { + let mut items = Vec::new(); + let mut token: Option = None; + loop { + let (page, next, truncated) = heal_storage + .list_objects_for_heal_page(bucket, "", token.as_deref()) + .await + .expect("b5 list page failed"); + items.extend(page); + if !truncated { + break; + } + token = next; + if token.is_none() { + break; + } + } + items +} + +/// Enumerate every version via the disk-walk UNION enumerator (walks all pages). +async fn enumerate_disk_walk(heal_storage: &Arc, bucket: &str) -> Vec { + let mut items = Vec::new(); + let mut token: Option = None; + loop { + let (page, next, truncated) = heal_storage + .list_versions_for_heal_page_disk_walk(SET_DISK_ID, bucket, "", token.as_deref()) + .await + .expect("disk-walk list page failed"); + items.extend(page); + if !truncated { + break; + } + token = next; + if token.is_none() { + break; + } + } + items +} + +mod serial_tests { + use super::*; + + /// The enumeration gap: a version surviving on only 1/4 disks (below the + /// read-quorum of 2) is OMITTED by the B5 `list_object_versions` enumeration + /// but INCLUDED by the disk-walk union enumerator. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn disk_walk_page_enumerates_subquorum_version_omitted_by_list_object_versions() { + let (disk_paths, ecstore, heal_storage) = setup_test_env_n(4).await; + let bucket = "b920-enum-gap"; + let object = "obj.bin"; + create_versioned_bucket(&ecstore, bucket).await; + + let v1 = put_versioned(&ecstore, bucket, object, &versioned_test_data(1)).await; + + // Wipe the object entirely on disks 1..4, leaving it on ONLY disk[0] + // (1/4 disks < read-quorum 2). effective listing_quorum for 4 drives is + // drives/2 = 2, so B5 cannot surface a 1-of-4 version. + for disk in &disk_paths[1..] { + let dir = object_dir(disk, bucket, object); + std::fs::remove_dir_all(&dir).expect("wipe object dir"); + } + assert!(xl_meta_path(&object_dir(&disk_paths[0], bucket, object)).exists()); + + let b5 = enumerate_b5(&heal_storage, bucket).await; + assert!( + !b5.iter().any(|it| it.version_id.as_deref() == Some(v1.as_str())), + "B5 read-quorum enumeration MUST omit the 1-of-4 sub-quorum version, got {b5:?}" + ); + + let walk = enumerate_disk_walk(&heal_storage, bucket).await; + assert!( + walk.iter() + .any(|it| it.name == object && it.version_id.as_deref() == Some(v1.as_str())), + "disk-walk union enumeration MUST include the 1-of-4 sub-quorum version, got {walk:?}" + ); + } + + /// DECISION 1 PIN (highest risk): xl.meta deleted on > parity disks while the + /// data shards remain. A Deep heal (grace disabled) must REGENERATE the lost + /// xl.meta and read back byte-identical — NOT dangling-delete the version. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn union_meta_lost_data_present_is_repaired_not_destroyed() { + disable_dangling_grace(); + let (disk_paths, ecstore, heal_storage) = setup_test_env_n(8).await; + let bucket = "b920-meta-lost"; + let object = "obj.bin"; + create_versioned_bucket(&ecstore, bucket).await; + + let data_v1 = versioned_test_data(7); + let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; + + // EC4+4: parity = 4. Delete ONLY xl.meta on 5 disks (> parity), leaving the + // data shards on all 8. Meta quorum (4) is now unreachable (3 metas), which + // WITHOUT the guard drives delete_if_dangling -> destruction (grace=0). + for disk in &disk_paths[0..5] { + remove_xl_meta_only(disk, bucket, object); + assert!(!xl_meta_path(&object_dir(disk, bucket, object)).exists()); + } + + // Heal the version through the real heal storage (Deep). + let (_result, error) = heal_storage + .heal_object(bucket, object, Some(&v1), &deep_heal_opts()) + .await + .expect("heal_object call must not itself error"); + assert!( + error.is_none(), + "recoverable version must heal without error (must NOT be dangling-deleted): {error:?}" + ); + + // xl.meta physically regenerated on the 5 meta-wiped disks. + for disk in &disk_paths[0..5] { + assert!( + xl_meta_path(&object_dir(disk, bucket, object)).exists(), + "xl.meta must be regenerated on the meta-wiped disk {disk:?}" + ); + } + + // The version is still present (NOT deleted) and reads back byte-identical. + assert_eq!( + read_version(&ecstore, bucket, object, &v1).await, + data_v1, + "rescued version must read back byte-identical" + ); + } + + /// Torn minority: a version whose DATA physically survives on FEWER than + /// data_blocks disks is genuinely unrecoverable. With grace disabled it must + /// still be dangling-deleted (the delete path is LIVE and the guard is + /// discriminating — it does not resurrect torn writes). + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn deep_heal_torn_minority_is_dangling_deleted_with_grace_zero() { + disable_dangling_grace(); + let (disk_paths, ecstore, heal_storage) = setup_test_env_n(4).await; + let bucket = "b920-torn"; + let object = "obj.bin"; + create_versioned_bucket(&ecstore, bucket).await; + + let v1 = put_versioned(&ecstore, bucket, object, &versioned_test_data(3)).await; + + // EC2+2 (4 drives, parity 2, data_blocks 2). Wipe the object ENTIRELY + // (meta + data) on 3 of 4 disks, leaving it on only 1 (< data_blocks 2): + // genuinely unrecoverable => dangling delete after (zero) grace. + for disk in &disk_paths[1..] { + std::fs::remove_dir_all(object_dir(disk, bucket, object)).expect("wipe object dir"); + } + // The surviving minority copy IS present before heal (so the delete below + // is a real destructive action, not a no-op). + assert_eq!(count_part_files(&object_dir(&disk_paths[0], bucket, object)), 1); + + let (_result, error) = heal_storage + .heal_object(bucket, object, Some(&v1), &deep_heal_opts()) + .await + .expect("heal_object call must not itself error"); + // A dangling delete reports the version as gone (FileVersionNotFound), + // proving the destructive path fired for a genuinely torn write. + assert!(error.is_some(), "a torn (< data_blocks) version must NOT be silently treated as healed"); + // Destructive-path PROOF: the stale minority copy on disk0 was purged by + // the dangling delete (the guard correctly did NOT rescue a torn write). + assert_eq!( + count_part_files(&object_dir(&disk_paths[0], bucket, object)), + 0, + "torn-write dangling delete must purge the surviving minority shard on disk0" + ); + // And the wiped disks are not resurrected. + for disk in &disk_paths[1..] { + assert_eq!( + count_part_files(&object_dir(disk, bucket, object)), + 0, + "torn-write heal must not resurrect data on the wiped disk {disk:?}" + ); + } + } + + /// A version present on exactly data_blocks=4 disks (< listing_quorum) yet + /// EC-reconstructable: after Deep heal the part.* + xl.meta are physically + /// restored on the 4 wiped disks and the data reads back byte-identical. B5 + /// omits it; the disk-walk enumerates it. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn deep_heal_restores_subquorum_but_reconstructable_version_wider_set() { + disable_dangling_grace(); + let (disk_paths, ecstore, heal_storage) = setup_test_env_n(8).await; + let bucket = "b920-reconstruct"; + let object = "obj.bin"; + create_versioned_bucket(&ecstore, bucket).await; + + let data_v1 = versioned_test_data(9); + let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; + + // EC4+4: wipe the object ENTIRELY on 4 disks, leaving full copies on the + // other 4 (== data_blocks). Meta quorum (4) still holds, so this heals via + // the normal reconstruction path once enumerated by the disk walk. + for disk in &disk_paths[0..4] { + std::fs::remove_dir_all(object_dir(disk, bucket, object)).expect("wipe object dir"); + assert_eq!(count_part_files(&object_dir(disk, bucket, object)), 0); + } + + // The disk-walk enumerates it (union view). + let walk = enumerate_disk_walk(&heal_storage, bucket).await; + assert!( + walk.iter().any(|it| it.version_id.as_deref() == Some(v1.as_str())), + "disk-walk must enumerate the reconstructable sub-quorum version" + ); + + let (_result, error) = heal_storage + .heal_object(bucket, object, Some(&v1), &deep_heal_opts()) + .await + .expect("heal_object call must not itself error"); + assert!(error.is_none(), "reconstructable version must heal cleanly: {error:?}"); + + // part.* + xl.meta physically restored on the 4 wiped disks. + for disk in &disk_paths[0..4] { + assert!( + xl_meta_path(&object_dir(disk, bucket, object)).exists(), + "xl.meta restored on wiped disk {disk:?}" + ); + assert_eq!( + count_part_files(&object_dir(disk, bucket, object)), + 1, + "data shard restored on wiped disk {disk:?}" + ); + } + assert_eq!( + read_version(&ecstore, bucket, object, &v1).await, + data_v1, + "reconstructed version must read back byte-identical" + ); + } + + /// The bounded disk-walk paginates across a multi-object bucket with + /// batch_objects=2, healing every version exactly once with a monotonic, + /// `dw1:`-tagged resume cursor and no anti-loop trip on the de-overlapped tail. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn disk_walk_multipage_resume_heals_each_version_once() { + let (_disk_paths, ecstore, _heal_storage) = setup_test_env_n(4).await; + let bucket = "b920-multipage"; + create_versioned_bucket(&ecstore, bucket).await; + + // 5 objects, 1 version each. + let mut expected: Vec<(String, String)> = Vec::new(); + for i in 0..5u8 { + let object = format!("obj-{i}.bin"); + let v = put_versioned(&ecstore, bucket, &object, &versioned_test_data(i + 20)).await; + expected.push((object, v)); + } + + // Page the raw ecstore disk-walk directly with an explicit small bound so + // pagination is actually exercised (the storage-API wrapper uses a large + // batch that would not split this fixture). + let mut seen: Vec<(String, Option)> = Vec::new(); + let mut forward: Option = None; + let mut pages = 0usize; + loop { + let (versions, next_forward, truncated) = ecstore + .heal_walk_versions_page(0, 0, bucket, "", forward.as_deref(), 2, 100_000) + .await + .expect("heal_walk_versions_page failed"); + pages += 1; + for v in &versions { + seen.push((v.name.clone(), v.version_id.clone())); + } + if !truncated { + break; + } + let nf = next_forward.expect("truncated page must carry a next_forward"); + // Cursor must strictly advance (monotonic) to avoid loops. + if let Some(prev) = &forward { + assert!(&nf > prev, "resume cursor must advance monotonically: {prev} -> {nf}"); + } + forward = Some(nf); + assert!(pages < 20, "pagination must terminate"); + } + assert!(pages >= 2, "batch_objects=2 over 5 objects must span multiple pages, pages={pages}"); + + // Every version enumerated EXACTLY once (no drops, no duplicates from the + // inclusive-forward de-overlap). + assert_eq!(seen.len(), expected.len(), "each version must appear exactly once: {seen:?}"); + for (object, v) in &expected { + let hits = seen + .iter() + .filter(|(n, vid)| n == object && vid.as_deref() == Some(v.as_str())) + .count(); + assert_eq!(hits, 1, "version {object}/{v} must be enumerated exactly once, got {hits}"); + } + } + + /// A disk that lost this object (its shard is gone) during the walk must not + /// cause a dangling delete: the union still enumerates the version from the + /// remaining disks (min_disks=1), and the still-quorum-present object heals + /// cleanly and reads back byte-identical (grace disabled). + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn offline_disk_during_walk_does_not_dangling_delete() { + disable_dangling_grace(); + let (disk_paths, ecstore, heal_storage) = setup_test_env_n(4).await; + let bucket = "b920-offline"; + let object = "obj.bin"; + create_versioned_bucket(&ecstore, bucket).await; + + let data_v1 = versioned_test_data(11); + let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; + + // Drop the object entirely on ONE disk (its shard + meta gone), leaving it + // on 3/4 (>= data_blocks 2). The union must still enumerate it. + std::fs::remove_dir_all(object_dir(&disk_paths[3], bucket, object)).expect("wipe object on disk3"); + + let walk = enumerate_disk_walk(&heal_storage, bucket).await; + assert!( + walk.iter().any(|it| it.version_id.as_deref() == Some(v1.as_str())), + "a disk missing this object must not drop the version from the union, got {walk:?}" + ); + + // Healing must NOT dangling-delete: the object is present on 3/4 (quorum). + // (Normal scan: the disk-walk enumerator gates on Deep OR AutoHeal, but the + // per-version repair itself is scan-mode agnostic — this pins that a + // missing-shard disk is reconstructed, not dangling-deleted.) + let normal_opts = HealOpts { + recreate: true, + remove: false, + ..Default::default() + }; + let (_result, error) = heal_storage + .heal_object(bucket, object, Some(&v1), &normal_opts) + .await + .expect("heal_object call must not itself error"); + assert!(error.is_none(), "quorum-present object must not be destroyed: {error:?}"); + assert!( + xl_meta_path(&object_dir(&disk_paths[3], bucket, object)).exists(), + "the missing shard's xl.meta must be restored on disk3" + ); + assert_eq!( + read_version(&ecstore, bucket, object, &v1).await, + data_v1, + "version must remain byte-identical after heal" + ); + } +}