diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 5c4b1474f..f7fe5c82f 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -19,7 +19,7 @@ use futures::future::join_all; use metrics::counter; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof}; use std::{ - collections::VecDeque, + collections::{HashSet, VecDeque}, future::Future, pin::Pin, sync::{Arc, OnceLock}, @@ -41,6 +41,33 @@ pub type PartialFn = Box]) -> Pin + Send>> + Send + 'static>; type FinishedFn = Box]) -> Pin + Send>> + Send + 'static>; +#[derive(Clone, Default)] +pub(crate) struct FallbackClaimTracker { + claimed: Arc>>, +} + +impl FallbackClaimTracker { + pub(crate) async fn claim_disk(&self, disk: &DiskStore) { + self.claimed.lock().await.insert(disk.endpoint().to_string()); + } + + pub(crate) async fn claimed_keys(&self) -> HashSet { + self.claimed.lock().await.clone() + } + + #[cfg(test)] + pub(crate) async fn claim_test_fallback(&self) { + let mut claimed = self.claimed.lock().await; + let key = format!("test-fallback-{}", claimed.len()); + claimed.insert(key); + } + + #[cfg(test)] + pub(crate) async fn contains_key(&self, key: &str) -> bool { + self.claimed.lock().await.contains(key) + } +} + #[derive(Debug)] enum PeekOutcome { Ready(Option), @@ -129,9 +156,28 @@ impl Clone for ListPathRawOptions { } pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> { + list_path_raw_inner(rx, opts, None).await +} + +pub(crate) async fn list_path_raw_with_claim_tracker( + rx: CancellationToken, + opts: ListPathRawOptions, + claim_tracker: FallbackClaimTracker, +) -> disk::error::Result<()> { + list_path_raw_inner(rx, opts, Some(claim_tracker)).await +} + +async fn list_path_raw_inner( + rx: CancellationToken, + opts: ListPathRawOptions, + fallback_claim_tracker: Option, +) -> disk::error::Result<()> { if opts.disks.is_empty() { return Err(DiskError::ErasureReadQuorum); } + if opts.min_disks > opts.disks.len() { + return Err(DiskError::ErasureReadQuorum); + } let log_bucket = opts.bucket.clone(); let log_path = opts.path.clone(); @@ -151,6 +197,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d for (disk_idx, disk) in opts.disks.iter().enumerate() { let opdisk = disk.clone(); let opts_clone = opts.clone(); + let fallback_claim_tracker = fallback_claim_tracker.clone(); let fds_clone = fds.clone(); #[cfg(test)] let test_fallbacks_clone = test_fallbacks.clone(); @@ -276,6 +323,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d while need_fallback { #[cfg(test)] if let Some(behavior) = take_fallback_candidate(&test_fallbacks_clone).await { + if let Some(claim_tracker) = fallback_claim_tracker.as_ref() { + claim_tracker.claim_test_fallback().await; + } match behavior { TestReaderBehavior::Eof => { need_fallback = false; @@ -341,6 +391,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); }; + if let Some(claim_tracker) = fallback_claim_tracker.as_ref() { + claim_tracker.claim_disk(&disk).await; + } let fallback_walk_started = std::time::Instant::now(); match disk @@ -797,6 +850,22 @@ mod tests { assert_eq!(err, DiskError::ErasureReadQuorum); } + #[tokio::test] + async fn list_path_raw_rejects_impossible_min_disks() { + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None], + min_disks: 3, + ..Default::default() + }, + ) + .await + .expect_err("impossible listing quorum should fail before producing partial results"); + + assert_eq!(err, DiskError::ErasureReadQuorum); + } + #[test] fn missing_path_error_classification_excludes_actionable_failures() { assert!(is_missing_path_error(&DiskError::FileNotFound)); @@ -889,6 +958,27 @@ mod tests { assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &[2]); } + #[tokio::test] + async fn list_path_raw_records_claimed_fallback_candidates() { + let claim_tracker = FallbackClaimTracker::default(); + + list_path_raw_with_claim_tracker( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None], + min_disks: 1, + test_reader_behaviors: vec![TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound)], + test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], + ..Default::default() + }, + claim_tracker.clone(), + ) + .await + .expect("fallback producer should restore the single logical reader"); + + assert!(claim_tracker.contains_key("test-fallback-0").await); + } + #[tokio::test] async fn list_path_raw_returns_timeout_when_reader_stalls_before_completion() { let err = list_path_raw( diff --git a/crates/ecstore/src/metadata/set_disk.rs b/crates/ecstore/src/metadata/set_disk.rs index b25117723..3568171a7 100644 --- a/crates/ecstore/src/metadata/set_disk.rs +++ b/crates/ecstore/src/metadata/set_disk.rs @@ -13,6 +13,14 @@ // limitations under the License. use super::*; +use rustfs_utils::http; + +#[derive(Clone, Copy)] +struct FileInfoIdentityGroup { + hash: [u8; 32], + count: usize, + mod_time: Option, +} impl SetDisks { pub(super) fn all_not_found_metadata(errs: &[Option]) -> bool { @@ -338,6 +346,90 @@ impl SetDisks { (new_disk, mod_time, None) } + fn usable_fileinfo_count(parts_metadata: &[FileInfo], errs: &[Option]) -> (usize, bool) { + let mut has_read_error = false; + let mut usable_metadata = 0; + for (meta, err) in parts_metadata.iter().zip(errs.iter()) { + if err.is_some() { + has_read_error = true; + continue; + } + + if meta.is_valid() { + usable_metadata += 1; + } + } + + (usable_metadata, has_read_error) + } + + pub(super) fn latest_fileinfo_selection_quorum( + version_id: &str, + parts_metadata: &[FileInfo], + errs: &[Option], + read_quorum: usize, + write_quorum: usize, + ) -> usize { + if !version_id.is_empty() || write_quorum <= read_quorum { + return read_quorum; + } + + let (usable_metadata, has_read_error) = Self::usable_fileinfo_count(parts_metadata, errs); + + if usable_metadata < write_quorum { + return read_quorum; + } + + if !has_read_error { + return write_quorum; + } + + let mut identity_counts = HashMap::with_capacity(usable_metadata); + for (meta, err) in parts_metadata.iter().zip(errs.iter()) { + if err.is_some() || !meta.is_valid() { + continue; + } + + let key = Self::file_info_quorum_hash(meta); + + let count = identity_counts.entry(key).or_insert(0); + *count += 1; + if *count >= write_quorum { + return write_quorum; + } + } + + read_quorum + } + + pub(super) fn select_valid_fileinfo( + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + version_id: &str, + read_quorum: usize, + write_quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo, usize)> { + let selection_quorum = + Self::latest_fileinfo_selection_quorum(version_id, parts_metadata, errs, read_quorum, write_quorum); + let (usable_metadata, has_read_error) = Self::usable_fileinfo_count(parts_metadata, errs); + + if version_id.is_empty() + && write_quorum > read_quorum + && has_read_error + && usable_metadata >= write_quorum + && selection_quorum == read_quorum + { + let (online_disks, fi) = Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_quorum)?; + return Ok((online_disks, fi, read_quorum)); + } + + let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, selection_quorum); + let fi = Self::pick_valid_fileinfo(parts_metadata, mod_time, etag, selection_quorum)?; + + Ok((online_disks, fi, selection_quorum)) + } + pub(super) fn pick_valid_fileinfo( metas: &[FileInfo], mod_time: Option, @@ -347,52 +439,271 @@ impl SetDisks { Self::find_file_info_in_quorum(metas, &mod_time, &etag, quorum) } + fn update_hash_bytes(hasher: &mut Sha256, value: &[u8]) { + hasher.update(value.len().to_le_bytes()); + hasher.update(value); + } + + fn update_hash_str(hasher: &mut Sha256, value: &str) { + Self::update_hash_bytes(hasher, value.as_bytes()); + } + + fn update_hash_optional_uuid(hasher: &mut Sha256, value: Option) { + if let Some(value) = value { + hasher.update([1]); + hasher.update(value.as_bytes()); + } else { + hasher.update([0]); + } + } + + fn update_hash_optional_time(hasher: &mut Sha256, value: Option) { + if let Some(value) = value { + hasher.update([1]); + hasher.update(value.unix_timestamp_nanos().to_le_bytes()); + } else { + hasher.update([0]); + } + } + + fn update_hash_optional_u32(hasher: &mut Sha256, value: Option) { + if let Some(value) = value { + hasher.update([1]); + hasher.update(value.to_le_bytes()); + } else { + hasher.update([0]); + } + } + + fn update_hash_optional_u64(hasher: &mut Sha256, value: Option) { + if let Some(value) = value { + hasher.update([1]); + hasher.update(value.to_le_bytes()); + } else { + hasher.update([0]); + } + } + + fn update_hash_optional_bytes(hasher: &mut Sha256, value: Option<&Bytes>) { + if let Some(value) = value { + hasher.update([1]); + Self::update_hash_bytes(hasher, value); + } else { + hasher.update([0]); + } + } + + fn update_hash_optional_str(hasher: &mut Sha256, value: Option<&str>) { + if let Some(value) = value { + hasher.update([1]); + Self::update_hash_str(hasher, value); + } else { + hasher.update([0]); + } + } + + fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool { + value + .get(..prefix.len()) + .is_some_and(|value_prefix| value_prefix.eq_ignore_ascii_case(prefix)) + } + + fn internal_metadata_suffix(name: &str) -> Option<&str> { + name.get(http::RUSTFS_INTERNAL_PREFIX.len()..) + .filter(|_| Self::starts_with_ignore_ascii_case(name, http::RUSTFS_INTERNAL_PREFIX)) + .or_else(|| { + name.get(http::MINIO_INTERNAL_PREFIX.len()..) + .filter(|_| Self::starts_with_ignore_ascii_case(name, http::MINIO_INTERNAL_PREFIX)) + }) + } + + fn is_replication_quorum_metadata_key(name: &str) -> bool { + if name.eq_ignore_ascii_case(http::AMZ_BUCKET_REPLICATION_STATUS) { + return true; + } + + let Some(suffix) = Self::internal_metadata_suffix(name) else { + return false; + }; + + suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICA_STATUS) + || suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICA_TIMESTAMP) + || suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICATION_STATUS) + || suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICATION_TIMESTAMP) + || suffix.eq_ignore_ascii_case(http::SUFFIX_PURGESTATUS) + || Self::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_RESET_ARN_PREFIX) + } + + fn update_hash_quorum_metadata_map(hasher: &mut Sha256, entries: &HashMap) { + let mut entries = entries + .iter() + .filter(|(name, _)| !Self::is_replication_quorum_metadata_key(name)) + .collect::>(); + entries.sort_by(|left, right| left.0.cmp(right.0)); + hasher.update(entries.len().to_le_bytes()); + for (name, value) in entries { + Self::update_hash_str(hasher, name); + Self::update_hash_str(hasher, value); + } + } + + fn file_info_quorum_hash(meta: &FileInfo) -> [u8; 32] { + let mut hasher = Sha256::new(); + Self::update_file_info_quorum_hash(&mut hasher, meta); + let digest = hasher.finalize(); + let mut key = [0u8; 32]; + key.copy_from_slice(digest.as_slice()); + key + } + fn update_file_info_quorum_hash(hasher: &mut Sha256, meta: &FileInfo) { hasher.update(meta.size.to_le_bytes()); - hasher.update([meta.deleted as u8, meta.mark_deleted as u8]); + hasher.update([u8::from(meta.deleted), u8::from(meta.mark_deleted)]); + hasher.update([u8::from(meta.expire_restored)]); + Self::update_hash_optional_time(hasher, meta.mod_time); + Self::update_hash_str(hasher, &meta.transition_status); + Self::update_hash_str(hasher, &meta.transition_tier); + Self::update_hash_str(hasher, &meta.transitioned_objname); + Self::update_hash_optional_uuid(hasher, meta.transition_version_id); + Self::update_hash_optional_u32(hasher, meta.mode); + Self::update_hash_optional_u64(hasher, meta.written_by_version); - if let Some(version_id) = meta.version_id { - hasher.update(version_id.as_bytes()); - } + Self::update_hash_optional_uuid(hasher, meta.version_id); + Self::update_hash_optional_uuid(hasher, meta.data_dir); - if let Some(data_dir) = meta.data_dir { - hasher.update(data_dir.as_bytes()); - } + Self::update_hash_optional_bytes(hasher, meta.checksum.as_ref()); - if let Some(checksum) = &meta.checksum { - hasher.update(checksum); - } + Self::update_hash_quorum_metadata_map(hasher, &meta.metadata); + hasher.update(meta.parts.len().to_le_bytes()); for part in meta.parts.iter() { - hasher.update(format!("part.{}", part.number).as_bytes()); - hasher.update(format!("part.{}", part.size).as_bytes()); + hasher.update(part.number.to_le_bytes()); + hasher.update(part.size.to_le_bytes()); hasher.update(part.actual_size.to_le_bytes()); - hasher.update(part.etag.as_bytes()); + Self::update_hash_str(hasher, &part.etag); - if let Some(mod_time) = part.mod_time { - hasher.update(mod_time.unix_timestamp_nanos().to_le_bytes()); - } + Self::update_hash_optional_time(hasher, part.mod_time); - if let Some(index) = &part.index { - hasher.update(index); - } + Self::update_hash_optional_bytes(hasher, part.index.as_ref()); + Self::update_hash_optional_str(hasher, part.error.as_deref()); if let Some(checksums) = &part.checksums { let mut checksum_entries = checksums.iter().collect::>(); checksum_entries.sort_by(|left, right| left.0.cmp(right.0)); + hasher.update(checksum_entries.len().to_le_bytes()); for (name, value) in checksum_entries { - hasher.update(name.as_bytes()); - hasher.update(value.as_bytes()); + Self::update_hash_str(hasher, name); + Self::update_hash_str(hasher, value); } + } else { + hasher.update(0usize.to_le_bytes()); } } if !meta.deleted && meta.size != 0 { - hasher.update(format!("{}+{}", meta.erasure.data_blocks, meta.erasure.parity_blocks).as_bytes()); - hasher.update(format!("{:?}", meta.erasure.distribution).as_bytes()); + hasher.update(meta.erasure.data_blocks.to_le_bytes()); + hasher.update(meta.erasure.parity_blocks.to_le_bytes()); + hasher.update(meta.erasure.distribution.len().to_le_bytes()); + for disk_index in meta.erasure.distribution.iter() { + hasher.update(disk_index.to_le_bytes()); + } } } + fn latest_fileinfo_identity_groups(parts_metadata: &[FileInfo], errs: &[Option]) -> Vec { + let mut groups: Vec = Vec::with_capacity(parts_metadata.len()); + for (meta, err) in parts_metadata.iter().zip(errs.iter()) { + if err.is_some() || !meta.is_valid() { + continue; + } + + let hash = Self::file_info_quorum_hash(meta); + if let Some(group) = groups.iter_mut().find(|group| group.hash == hash) { + group.count += 1; + continue; + } + + groups.push(FileInfoIdentityGroup { + hash, + count: 1, + mod_time: meta.mod_time, + }); + } + + groups + } + + fn pick_fileinfo_identity( + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + hash: [u8; 32], + quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo)> { + let mut online_disks = vec![None; disks.len()]; + let mut selected = None; + let mut count = 0; + + for (i, ((meta, err), disk)) in parts_metadata.iter().zip(errs.iter()).zip(disks.iter()).enumerate() { + if err.is_some() || !meta.is_valid() || Self::file_info_quorum_hash(meta) != hash { + continue; + } + + count += 1; + online_disks[i].clone_from(disk); + if selected.is_none() { + selected = Some(meta.clone()); + } + } + + if count < quorum { + return Err(DiskError::ErasureReadQuorum); + } + + selected + .map(|mut fi| { + fi.is_latest = fi.successor_mod_time.is_none(); + (online_disks, fi) + }) + .ok_or(DiskError::ErasureReadQuorum) + } + + fn pick_degraded_latest_fileinfo( + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + read_quorum: usize, + write_quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo)> { + let mut groups = Self::latest_fileinfo_identity_groups(parts_metadata, errs); + if groups.is_empty() { + return Err(DiskError::ErasureReadQuorum); + } + + groups.sort_by(|left, right| right.mod_time.cmp(&left.mod_time).then_with(|| right.count.cmp(&left.count))); + let latest_mod_time = groups[0].mod_time; + + let mut older_start = 0; + while older_start < groups.len() && groups[older_start].mod_time == latest_mod_time { + if groups[older_start].count >= write_quorum { + return Self::pick_fileinfo_identity(disks, parts_metadata, errs, groups[older_start].hash, write_quorum); + } + older_start += 1; + } + + if older_start > 1 { + return Err(DiskError::ErasureReadQuorum); + } + + for group in groups.iter().skip(older_start) { + if group.count >= read_quorum { + return Self::pick_fileinfo_identity(disks, parts_metadata, errs, group.hash, read_quorum); + } + } + + Err(DiskError::ErasureReadQuorum) + } + pub(super) fn find_file_info_in_quorum( metas: &[FileInfo], mod_time: &Option, @@ -405,7 +716,6 @@ impl SetDisks { } let mut meta_hashes = vec![None; metas.len()]; - let mut hasher = Sha256::new(); for (i, meta) in metas.iter().enumerate() { if !meta.is_valid() { @@ -437,8 +747,6 @@ impl SetDisks { let mod_valid = mod_time == &meta.mod_time; if etag_only || mod_valid { - Self::update_file_info_quorum_hash(&mut hasher, meta); - if meta.is_remote() { // TODO: } @@ -447,9 +755,7 @@ impl SetDisks { // TODO: IsCompressed - meta_hashes[i] = Some(hex(hasher.clone().finalize().as_slice())); - - hasher.reset(); + meta_hashes[i] = Some(Self::file_info_quorum_hash(meta)); } else { debug!( index = i, @@ -462,7 +768,7 @@ impl SetDisks { let mut count_map = HashMap::new(); - for hash in meta_hashes.iter().flatten() { + for hash in meta_hashes.iter().flatten().copied() { *count_map.entry(hash).or_insert(0) += 1; } @@ -495,7 +801,7 @@ impl SetDisks { for (i, op_hash) in meta_hashes.iter().enumerate() { if let Some(hash) = op_hash && let Some(max_hash) = max_val - && hash == max_hash + && *hash == max_hash && metas[i].is_valid() { if !found { @@ -504,7 +810,7 @@ impl SetDisks { } let props = ObjProps { - mod_time: metas[i].mod_time, + successor_mod_time: metas[i].successor_mod_time, num_versions: metas[i].num_versions, }; @@ -517,9 +823,9 @@ impl SetDisks { for (val, &count) in &valid_obj_map { if count >= quorum { - fi.mod_time = val.mod_time; + fi.successor_mod_time = val.successor_mod_time; fi.num_versions = val.num_versions; - fi.is_latest = val.mod_time.is_none(); + fi.is_latest = val.successor_mod_time.is_none(); break; } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 9945aff8f..35165e133 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -97,7 +97,8 @@ use rustfs_common::heal_channel::{ use rustfs_config::MI_B; use rustfs_filemeta::{ FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo, - RawFileInfo, ReplicateDecision, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions, + RawFileInfo, ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, + merge_file_meta_versions, }; use rustfs_io_metrics::{ record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration, @@ -3475,7 +3476,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { }; let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) { - Ok((r, w)) => (r as usize, w as usize), + Ok((r, w)) => ( + usize::try_from(r) + .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?, + usize::try_from(w) + .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?, + ), Err(mut err) => { if err == DiskError::ErasureReadQuorum && !src_bucket.starts_with(RUSTFS_META_BUCKET) @@ -3494,10 +3500,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } }; - let (online_disks, mod_time, etag) = Self::list_online_disks(&disks, &metas, &errs, read_quorum); - - let mut fi = Self::pick_valid_fileinfo(&metas, mod_time, etag, read_quorum) - .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; + let src_version_id = src_opts.version_id.as_deref().unwrap_or_default(); + let (online_disks, mut fi, _) = + Self::select_valid_fileinfo(&disks, &metas, &errs, src_version_id, read_quorum, write_quorum) + .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; if fi.deleted { if src_opts.version_id.is_none() { @@ -4078,8 +4084,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } }; - let read_quorum = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) { - Ok((res, _)) => res, + let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) { + Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum), Err(mut err) => { if err == DiskError::ErasureReadQuorum && !bucket.starts_with(RUSTFS_META_BUCKET) @@ -4098,11 +4104,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } }; - let read_quorum = read_quorum as usize; + let read_quorum = + usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?; + let write_quorum = usize::try_from(write_quorum) + .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?; - let (online_disks, mod_time, etag) = Self::list_online_disks(&disks, &metas, &errs, read_quorum); - - let mut fi = Self::pick_valid_fileinfo(&metas, mod_time, etag, read_quorum) + let version_id = opts.version_id.as_deref().unwrap_or_default(); + let (online_disks, mut fi, _) = Self::select_valid_fileinfo(&disks, &metas, &errs, version_id, read_quorum, write_quorum) .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; if fi.deleted { @@ -6045,13 +6053,13 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { #[derive(Debug, PartialEq, Eq)] struct ObjProps { - mod_time: Option, + successor_mod_time: Option, num_versions: usize, } impl Hash for ObjProps { fn hash(&self, state: &mut H) { - self.mod_time.hash(state); + self.successor_mod_time.hash(state); self.num_versions.hash(state); } } @@ -8062,6 +8070,34 @@ mod tests { } } + fn decoded_quorum_test_fileinfo_with_metadata( + mod_time: OffsetDateTime, + data_dir: Uuid, + part_etag: &str, + erasure_index: usize, + extra_metadata: &[(&str, &str)], + ) -> FileInfo { + let mut fi = quorum_test_fileinfo(mod_time, data_dir, part_etag, erasure_index); + for (name, value) in extra_metadata { + fi.metadata.insert((*name).to_string(), (*value).to_string()); + } + + 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"); + rustfs_filemeta::get_file_info( + &encoded, + "bucket", + "object", + "", + rustfs_filemeta::FileInfoOpts { + data: false, + include_free_versions: false, + }, + ) + .expect("test file metadata should decode as file info") + } + #[test] fn test_find_file_info_in_quorum_uses_part_identity() { let mod_time = OffsetDateTime::now_utc(); @@ -8096,6 +8132,319 @@ mod tests { assert_eq!(err, DiskError::ErasureReadQuorum); } + #[test] + fn test_latest_fileinfo_selection_quorum_requires_write_quorum_when_full_metadata_is_available() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 3), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 4), + ]; + let errs = vec![None, None, None, None]; + + let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3); + + assert_eq!(quorum, 3); + } + + #[test] + fn test_latest_fileinfo_selection_quorum_preserves_read_quorum_for_version_or_degraded_reads() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2), + FileInfo::default(), + FileInfo::default(), + ]; + let degraded_errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)]; + let clean_errs = vec![None, None, None, None]; + + assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, °raded_errs, 2, 3), 2); + assert_eq!(SetDisks::latest_fileinfo_selection_quorum("version-id", &metas, &clean_errs, 2, 3), 2); + } + + #[test] + fn test_latest_fileinfo_selection_quorum_keeps_read_quorum_for_partial_overwrite_with_read_error() { + let old_mod_time = OffsetDateTime::now_utc(); + let new_mod_time = old_mod_time + time::Duration::seconds(1); + let old_data_dir = Uuid::new_v4(); + let new_data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", 1), + quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", 2), + quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", 3), + FileInfo::default(), + ]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3); + let (online_disks, mod_time, etag) = SetDisks::list_online_disks(&vec![None; metas.len()], &metas, &errs, quorum); + let fi = SetDisks::pick_valid_fileinfo(&metas, mod_time, etag, quorum) + .expect("old metadata should remain readable with read quorum"); + + assert_eq!(quorum, 2); + assert_eq!(online_disks.len(), metas.len()); + assert_eq!(fi.data_dir, Some(old_data_dir)); + assert_eq!(fi.parts[0].etag, "part-etag-old"); + + let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3) + .expect("old metadata should remain selectable with read quorum"); + assert_eq!(selected_quorum, 2); + assert_eq!(selected.data_dir, Some(old_data_dir)); + assert_eq!(selected.parts[0].etag, "part-etag-old"); + assert!(selected.is_latest); + } + + #[test] + fn test_latest_fileinfo_selection_rejects_partial_latest_read_quorum_with_read_error() { + let old_mod_time = OffsetDateTime::now_utc(); + let new_mod_time = old_mod_time + time::Duration::seconds(1); + let old_data_dir = Uuid::new_v4(); + let new_data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", 1), + quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", 2), + quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", 3), + FileInfo::default(), + ]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + + assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + } + + #[test] + fn test_latest_fileinfo_selection_preserves_degraded_read_quorum_without_competing_latest() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(mod_time, data_dir, "part-etag-old", 1), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-old", 2), + FileInfo::default(), + FileInfo::default(), + ]; + let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)]; + + let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3) + .expect("read quorum should remain enough when no competing latest is visible"); + + assert_eq!(selected_quorum, 2); + assert_eq!(selected.data_dir, Some(data_dir)); + assert_eq!(selected.parts[0].etag, "part-etag-old"); + } + + #[test] + fn test_latest_fileinfo_selection_ignores_derived_version_stack_drift() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let mut latest_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1); + latest_meta.is_latest = true; + latest_meta.num_versions = 1; + + let mut stale_stack_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2); + stale_stack_meta.is_latest = false; + stale_stack_meta.successor_mod_time = Some(mod_time + time::Duration::seconds(1)); + stale_stack_meta.num_versions = 2; + + let mut newer_stack_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3); + newer_stack_meta.is_latest = false; + newer_stack_meta.successor_mod_time = Some(mod_time + time::Duration::seconds(2)); + newer_stack_meta.num_versions = 3; + + let metas = vec![latest_meta, stale_stack_meta, newer_stack_meta, FileInfo::default()]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3) + .expect("same object version should stay readable despite derived version stack drift"); + + assert_eq!(selected_quorum, 3); + assert_eq!(selected.data_dir, Some(data_dir)); + assert_eq!(selected.parts[0].etag, "part-etag"); + assert_eq!(selected.mod_time, Some(mod_time)); + } + + #[test] + fn test_latest_fileinfo_selection_uses_successor_mod_time_quorum_for_latest_flag() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let mut stale_stack_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1); + stale_stack_meta.is_latest = false; + stale_stack_meta.successor_mod_time = Some(mod_time + time::Duration::seconds(1)); + stale_stack_meta.num_versions = 2; + + let mut latest_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2); + latest_meta_a.is_latest = true; + latest_meta_a.num_versions = 1; + let mut latest_meta_b = latest_meta_a.clone(); + latest_meta_b.erasure.index = 3; + + let metas = vec![stale_stack_meta, latest_meta_a, latest_meta_b]; + + let selected = SetDisks::find_file_info_in_quorum(&metas, &Some(mod_time), &None, 2) + .expect("latest flag should be derived from successor mod time quorum"); + + assert!(selected.is_latest); + assert_eq!(selected.successor_mod_time, None); + assert_eq!(selected.num_versions, 1); + assert_eq!(selected.mod_time, Some(mod_time)); + } + + #[test] + fn test_latest_fileinfo_selection_ignores_replication_state_drift() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let replication_status_key = format!( + "{}{}", + rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::SUFFIX_REPLICATION_STATUS + ); + let replication_timestamp_key = format!( + "{}{}", + rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP + ); + let replication_reset_key = format!( + "{}{}target-a", + rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::SUFFIX_REPLICATION_RESET_ARN_PREFIX + ); + let meta_a = decoded_quorum_test_fileinfo_with_metadata( + mod_time, + data_dir, + "part-etag", + 1, + &[ + (&replication_status_key, "target-a=COMPLETED;"), + (&replication_timestamp_key, "2024-01-01T00:00:00Z"), + (&replication_reset_key, "COMPLETED"), + ], + ); + let meta_b = decoded_quorum_test_fileinfo_with_metadata( + mod_time, + data_dir, + "part-etag", + 2, + &[ + (&replication_status_key, "target-a=PENDING;"), + (&replication_timestamp_key, "2024-01-01T00:00:01Z"), + (&replication_reset_key, "PENDING"), + ], + ); + let meta_c = decoded_quorum_test_fileinfo_with_metadata( + mod_time, + data_dir, + "part-etag", + 3, + &[ + (&replication_status_key, "target-a=FAILED;"), + (&replication_timestamp_key, "2024-01-01T00:00:02Z"), + (&replication_reset_key, "FAILED"), + ], + ); + assert!(meta_a.replication_state_internal.is_some()); + assert_eq!( + meta_a + .metadata + .get(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS) + .map(String::as_str), + Some("COMPLETED") + ); + + let metas = vec![meta_a, meta_b, meta_c, FileInfo::default()]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3) + .expect("replication status drift should not split readable object identity"); + + assert_eq!(selected_quorum, 3); + assert_eq!(selected.data_dir, Some(data_dir)); + assert_eq!(selected.parts[0].etag, "part-etag"); + } + + #[test] + fn test_latest_fileinfo_selection_rejects_same_modtime_metadata_split_without_write_quorum() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let mut old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1); + let mut old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2); + let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3); + old_meta_a.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string()); + old_meta_b.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string()); + partial_meta + .metadata + .insert("x-amz-meta-color".to_string(), "red".to_string()); + let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3); + let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + + assert_eq!(quorum, 2); + assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + } + + #[test] + fn test_latest_fileinfo_selection_rejects_same_modtime_partial_metadata_read_quorum() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let mut old_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1); + let mut partial_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2); + let mut partial_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3); + old_meta.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string()); + partial_meta_a + .metadata + .insert("x-amz-meta-color".to_string(), "red".to_string()); + partial_meta_b + .metadata + .insert("x-amz-meta-color".to_string(), "red".to_string()); + let metas = vec![old_meta, partial_meta_a, partial_meta_b, FileInfo::default()]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + + assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + } + + #[test] + fn test_latest_fileinfo_selection_rejects_same_modtime_transition_split_without_write_quorum() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1); + let old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2); + let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3); + partial_meta.transition_status = TRANSITION_COMPLETE.to_string(); + partial_meta.transition_tier = "WARM".to_string(); + partial_meta.transitioned_objname = "remote/object".to_string(); + partial_meta.transition_version_id = Some(Uuid::new_v4()); + let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3); + let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + + assert_eq!(quorum, 2); + assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + } + + #[test] + fn test_latest_fileinfo_selection_quorum_uses_write_quorum_for_degraded_committed_identity() { + let mod_time = OffsetDateTime::now_utc(); + let data_dir = Uuid::new_v4(); + let metas = vec![ + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2), + quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 3), + FileInfo::default(), + ]; + let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; + + assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3), 3); + } + #[test] fn test_list_object_parities() { // Test extracting parity counts from file info diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 73cdc8620..628e4c2be 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2362,7 +2362,7 @@ impl SetDisks { let _min_disks = self.set_drive_count - self.default_parity_count; let metadata_resolve_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let (read_quorum, _) = match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count) + let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count) .map_err(|err| to_object_err(err.into(), vec![bucket, object])) { Ok(v) => v, @@ -2378,8 +2378,8 @@ impl SetDisks { }; let read_quorum = usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?; - metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, read_quorum); - + let write_quorum = usize::try_from(write_quorum) + .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?; if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { error!("reduce_read_quorum_errs: {:?}, bucket: {}, object: {}", &err, bucket, object); record_get_stage_duration_if_enabled( @@ -2390,9 +2390,9 @@ impl SetDisks { return Err(to_object_err(err.into(), vec![bucket, object])); } - let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum); - - let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum)?; + let (op_online_disks, fi, fileinfo_selection_quorum) = + Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum)?; + metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, fileinfo_selection_quorum); if errs.iter().any(|err| err.is_some()) { let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref()); submit_read_repair_heal( diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index b5bf00684..179a843ee 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -15,10 +15,10 @@ use crate::bucket::metadata_sys::get_versioning_config; use crate::bucket::utils::check_list_objs_args; use crate::bucket::versioning::VersioningApi; -use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; +use crate::cache_value::metacache_set::{FallbackClaimTracker, ListPathRawOptions, list_path_raw_with_claim_tracker}; use crate::core::sets::Sets; use crate::disk::error::DiskError; -use crate::disk::{DiskInfo, DiskStore}; +use crate::disk::{DiskAPI, DiskInfo, DiskStore, WalkDirOptions}; use crate::error::{ Error, Result, StorageError, is_all_not_found, is_all_volume_not_found, is_err_bucket_not_found, to_object_err, }; @@ -36,12 +36,14 @@ use crate::store::utils::is_reserved_or_invalid_bucket; use futures::future::join_all; use rand::seq::SliceRandom; use rustfs_filemeta::{ - MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntriesSortedResult, MetaCacheEntry, MetadataResolutionParams, - merge_file_meta_versions, + FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntriesSortedResult, MetaCacheEntry, + MetacacheReader, MetadataResolutionParams, is_io_eof, merge_file_meta_versions, }; use rustfs_utils::path::{self, SLASH_SEPARATOR, base_dir_from_prefix}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use tokio::io::duplex; +use tokio::sync::OnceCell; use tokio::sync::broadcast::{self}; use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio_util::sync::CancellationToken; @@ -408,10 +410,20 @@ fn list_objects_paginate( (objects, prefixes, is_truncated, next_marker, next_version_idmarker) } -fn list_metadata_resolution_params(bucket: String, listing_quorum: usize, versioned: bool) -> MetadataResolutionParams { +fn list_metadata_resolution_params( + bucket: String, + listing_quorum: usize, + latest_object_quorum: usize, + versioned: bool, +) -> MetadataResolutionParams { + let quorum = if versioned { + listing_quorum + } else { + latest_object_quorum.max(listing_quorum) + }; let mut resolver = MetadataResolutionParams { - dir_quorum: listing_quorum, - obj_quorum: listing_quorum, + dir_quorum: quorum, + obj_quorum: quorum, bucket, ..Default::default() }; @@ -423,6 +435,544 @@ fn list_metadata_resolution_params(bucket: String, listing_quorum: usize, versio resolver } +fn resolve_listing_entries( + entries: MetaCacheEntries, + resolver: MetadataResolutionParams, + enforce_write_quorum: bool, +) -> Option { + if enforce_write_quorum { + entries.resolve_with_write_quorum(resolver) + } else { + entries.resolve(resolver) + } +} + +enum ListingEntryResolution { + Resolved(MetaCacheEntry), + NeedsSupplement(MetaCacheEntry, Option), + Rejected, +} + +#[derive(Clone)] +struct FallbackListingEntry { + endpoint: String, + entry: MetaCacheEntry, +} + +type FallbackListingEntries = HashMap>; + +#[derive(Clone)] +struct ListingSupplementOptions { + bucket: String, + path: String, + recursive: bool, + filter_prefix: Option, + forward_to: Option, + per_disk_limit: i32, + skip_total_timeout: bool, +} + +struct ListingSupplement { + options: ListingSupplementOptions, + fallback_disks: Arc>, + claim_tracker: FallbackClaimTracker, + entries: Option>>, +} + +impl ListingSupplement { + fn new( + options: ListingSupplementOptions, + fallback_disks: Arc>, + claim_tracker: FallbackClaimTracker, + ) -> Arc { + let entries = if options.per_disk_limit > 0 { + Some(Arc::new(OnceCell::new())) + } else { + None + }; + + Arc::new(Self { + options, + fallback_disks, + claim_tracker, + entries, + }) + } + + fn is_empty(&self) -> bool { + self.fallback_disks.is_empty() + } + + async fn entries_for(&self, object: &str) -> Vec> { + if self.fallback_disks.is_empty() { + return Vec::new(); + } + + let claimed_keys = self.claim_tracker.claimed_keys().await; + let Some(entries) = &self.entries else { + return self.read_object_entries(object, &claimed_keys).await; + }; + + let entries = entries.get_or_init(|| self.load_entries()).await; + let claimed_keys = self.claim_tracker.claimed_keys().await; + fallback_entries_for_object(entries, object, &claimed_keys) + } + + async fn load_entries(&self) -> FallbackListingEntries { + let claimed_keys = self.claim_tracker.claimed_keys().await; + let futures = self.fallback_disks.iter().filter_map(|disk| { + let endpoint = disk.endpoint().to_string(); + if claimed_keys.contains(&endpoint) { + return None; + } + + Some(read_fallback_listing_disk(disk.clone(), endpoint, self.options.clone())) + }); + + let per_disk_entries = join_all(futures).await; + let entry_count = per_disk_entries.iter().map(Vec::len).sum(); + let mut entries = FallbackListingEntries::with_capacity(entry_count); + for disk_entries in per_disk_entries { + for entry in disk_entries { + entries.entry(entry.entry.name.clone()).or_default().push(entry); + } + } + + entries + } + + async fn read_object_entries(&self, object: &str, claimed_keys: &HashSet) -> Vec> { + let futures = self.fallback_disks.iter().filter_map(|disk| { + let endpoint = disk.endpoint().to_string(); + if claimed_keys.contains(&endpoint) { + return None; + } + + Some(read_fallback_object_disk( + disk.clone(), + endpoint, + self.options.bucket.clone(), + object.to_owned(), + )) + }); + + let entries = join_all(futures).await; + let claimed_keys = self.claim_tracker.claimed_keys().await; + entries + .into_iter() + .flatten() + .filter(|entry| !claimed_keys.contains(&entry.endpoint)) + .map(|entry| Some(entry.entry)) + .collect() + } +} + +fn fallback_entries_for_object( + entries: &FallbackListingEntries, + object: &str, + claimed_keys: &HashSet, +) -> Vec> { + entries + .get(object) + .into_iter() + .flat_map(|entries| entries.iter()) + .filter(|entry| !claimed_keys.contains(&entry.endpoint)) + .map(|entry| Some(entry.entry.clone())) + .collect() +} + +async fn read_fallback_listing_disk( + disk: DiskStore, + endpoint: String, + options: ListingSupplementOptions, +) -> Vec { + let (rd, mut wr) = duplex(64); + let walk_endpoint = endpoint.clone(); + let walk_job = tokio::spawn(async move { + disk.walk_dir( + WalkDirOptions { + bucket: options.bucket, + base_dir: options.path, + recursive: options.recursive, + report_notfound: false, + filter_prefix: options.filter_prefix, + forward_to: options.forward_to, + limit: options.per_disk_limit, + skip_total_timeout: options.skip_total_timeout, + ..Default::default() + }, + &mut wr, + ) + .await + }); + + let mut reader = MetacacheReader::new(rd); + let mut entries = Vec::new(); + let mut stream_ok = true; + loop { + match reader.peek().await { + Ok(Some(entry)) => { + entries.push(FallbackListingEntry { + endpoint: endpoint.clone(), + entry, + }); + if let Err(err) = reader.skip(1).await { + debug!( + endpoint = %endpoint, + error = ?err, + "fallback listing supplement stream failed while advancing reader" + ); + stream_ok = false; + break; + } + } + Ok(None) => break, + Err(err) if err == rustfs_filemeta::Error::Unexpected || is_io_eof(&err) => break, + Err(err) => { + debug!( + endpoint = %endpoint, + error = ?err, + "fallback listing supplement stream failed while reading entry" + ); + stream_ok = false; + break; + } + } + } + drop(reader); + + match walk_job.await { + Ok(Ok(())) if stream_ok => entries, + Ok(Ok(())) => Vec::new(), + Ok(Err(err)) => { + debug!( + endpoint = %walk_endpoint, + error = ?err, + "fallback listing supplement walk_dir failed" + ); + Vec::new() + } + Err(err) => { + debug!( + endpoint = %walk_endpoint, + error = ?err, + "fallback listing supplement task failed" + ); + Vec::new() + } + } +} + +async fn read_fallback_object_disk( + disk: DiskStore, + endpoint: String, + bucket: String, + object: String, +) -> Option { + match disk.read_xl(&bucket, &object, false).await { + Ok(raw) if raw.buf.is_empty() => None, + Ok(raw) => Some(FallbackListingEntry { + endpoint, + entry: MetaCacheEntry { + name: object, + metadata: raw.buf, + cached: None, + reusable: false, + }, + }), + Err(err) if DiskError::is_err_object_not_found(&err) || DiskError::is_err_version_not_found(&err) => None, + Err(err) => { + debug!( + endpoint = %endpoint, + bucket = %bucket, + object = %object, + error = ?err, + "fallback listing supplement read_xl failed" + ); + None + } + } +} + +fn version_requires_supplement( + version_required_quorum: usize, + reader_disks: usize, + selected_object_versions: usize, + requested_versions: usize, +) -> bool { + reader_disks < version_required_quorum && (requested_versions == 0 || selected_object_versions < requested_versions) +} + +fn cached_entry_needs_supplement( + cached: &FileMeta, + reader_disks: usize, + resolver: &MetadataResolutionParams, + enforce_write_quorum: bool, +) -> bool { + if !enforce_write_quorum { + return false; + } + + let mut selected_object_versions = 0; + for version in cached.versions.iter() { + let required_quorum = version.write_quorum(resolver.obj_quorum).max(resolver.obj_quorum); + if version_requires_supplement(required_quorum, reader_disks, selected_object_versions, resolver.requested_versions) { + return true; + } + + if !version.header.free_version() { + selected_object_versions += 1; + } + if resolver.requested_versions > 0 && selected_object_versions == resolver.requested_versions { + break; + } + } + + false +} + +fn listing_entries_supplement_target( + entries: &MetaCacheEntries, + resolver: &MetadataResolutionParams, + enforce_write_quorum: bool, +) -> Option { + if !enforce_write_quorum { + return None; + } + + for entry in entries.0.iter().flatten() { + if entry.is_dir() { + continue; + } + + let reader_disks = entries + .0 + .iter() + .filter(|candidate| { + candidate + .as_ref() + .is_some_and(|candidate| candidate.name == entry.name && candidate.is_object()) + }) + .count(); + let mut entry = entry.clone(); + if let Ok(cached) = entry.xl_meta() + && cached_entry_needs_supplement(&cached, reader_disks, resolver, enforce_write_quorum) + { + return Some(entry.name); + } + } + + None +} + +async fn resolve_listing_entries_with_supplement( + entries: MetaCacheEntries, + resolver: MetadataResolutionParams, + enforce_write_quorum: bool, + supplement: Arc, +) -> Option { + if !supplement.is_empty() + && let Some(object) = listing_entries_supplement_target(&entries, &resolver, enforce_write_quorum) + { + let mut candidates = entries.0; + candidates.extend(supplement.entries_for(&object).await); + return resolve_listing_entries(MetaCacheEntries(candidates), resolver, enforce_write_quorum); + } + + resolve_listing_entries(entries, resolver, enforce_write_quorum) +} + +async fn resolve_agreed_listing_entry_with_supplement( + entry: MetaCacheEntry, + reader_disks: usize, + resolver: MetadataResolutionParams, + enforce_write_quorum: bool, + supplement: Arc, +) -> Option { + if supplement.is_empty() { + return None; + } + + let mut candidates = Vec::with_capacity(reader_disks); + candidates.resize(reader_disks, Some(entry.clone())); + candidates.extend(supplement.entries_for(&entry.name).await); + resolve_listing_entries(MetaCacheEntries(candidates), resolver, enforce_write_quorum) +} + +fn listing_entry_with_selected_versions( + mut entry: MetaCacheEntry, + meta_ver: u8, + data: rustfs_filemeta::InlineData, + selected_versions: Vec, +) -> Option { + let merged_cached = FileMeta { + meta_ver, + data, + versions: selected_versions, + }; + let metadata = merged_cached.marshal_msg().ok()?; + + entry.metadata = metadata; + entry.cached = Some(merged_cached); + Some(entry) +} + +fn resolve_agreed_listing_entry( + mut entry: MetaCacheEntry, + reader_disks: usize, + resolver: MetadataResolutionParams, + enforce_write_quorum: bool, +) -> ListingEntryResolution { + if !enforce_write_quorum { + return ListingEntryResolution::Resolved(entry); + } + + if entry.is_dir() { + return if reader_disks >= resolver.dir_quorum { + ListingEntryResolution::Resolved(entry) + } else { + ListingEntryResolution::Rejected + }; + } + + if reader_disks < resolver.obj_quorum { + return ListingEntryResolution::Rejected; + } + + let cached = match entry.xl_meta() { + Ok(cached) => cached, + Err(_) if resolver.obj_quorum <= 1 => return ListingEntryResolution::Resolved(entry), + Err(_) => return ListingEntryResolution::Rejected, + }; + let mut selected_versions = Vec::new(); + let mut selected_object_versions = 0; + let mut first_selected_idx = None; + let mut needs_supplement = false; + + for (idx, version) in cached.versions.iter().enumerate() { + let required_quorum = version.write_quorum(resolver.obj_quorum).max(resolver.obj_quorum); + if reader_disks < required_quorum { + needs_supplement |= + version_requires_supplement(required_quorum, reader_disks, selected_object_versions, resolver.requested_versions); + continue; + } + + if first_selected_idx.is_none() { + first_selected_idx = Some(idx); + } + if !version.header.free_version() { + selected_object_versions += 1; + } + selected_versions.push(version.clone()); + + if resolver.requested_versions > 0 && selected_object_versions == resolver.requested_versions { + break; + } + } + + if needs_supplement { + let fallback = if selected_versions.is_empty() { + None + } else if selected_versions.len() == cached.versions.len() { + Some(entry.clone()) + } else { + listing_entry_with_selected_versions(entry.clone(), cached.meta_ver, cached.data, selected_versions) + }; + return ListingEntryResolution::NeedsSupplement(entry, fallback); + } + + if selected_versions.is_empty() { + return ListingEntryResolution::Rejected; + } + + let selected_latest = first_selected_idx == Some(0) + && resolver.requested_versions == 1 + && selected_versions + .first() + .is_some_and(|version| !version.header.free_version()); + if selected_latest || selected_versions.len() == cached.versions.len() { + return ListingEntryResolution::Resolved(entry); + } + + listing_entry_with_selected_versions(entry, cached.meta_ver, cached.data, selected_versions) + .map(ListingEntryResolution::Resolved) + .unwrap_or(ListingEntryResolution::Rejected) +} + +fn latest_listing_object_quorum( + listing_quorum: usize, + drive_count: usize, + parity_count: usize, + enforce_write_quorum: bool, +) -> usize { + latest_listing_required_object_quorum(listing_quorum, drive_count, parity_count, enforce_write_quorum) +} + +fn latest_listing_required_object_quorum( + listing_quorum: usize, + drive_count: usize, + parity_count: usize, + enforce_write_quorum: bool, +) -> usize { + if !enforce_write_quorum { + return listing_quorum; + } + + write_quorum_for_drive_count(drive_count, parity_count).max(listing_quorum) +} + +fn enforce_latest_listing_write_quorum(strict_latest: bool, ask_disks: &str) -> bool { + strict_latest && !matches!(normalize_list_quorum(ask_disks), "disk" | "reduced") +} + +#[cfg(test)] +fn latest_listing_allow_agreed_objects(enforce_write_quorum: bool, reader_disks: usize, object_quorum: usize) -> bool { + !enforce_write_quorum || reader_disks >= object_quorum +} + +fn latest_listing_raw_min_disks(listing_quorum: usize, object_quorum: usize, enforce_write_quorum: bool) -> usize { + if enforce_write_quorum { object_quorum } else { listing_quorum } +} + +fn positive_ask_disks(ask_disks: i32) -> Option { + usize::try_from(ask_disks).ok().filter(|asked| *asked > 0) +} + +fn listing_quorum_from_ask_disks(ask_disks: i32) -> usize { + positive_ask_disks(ask_disks) + .map(|asked| asked.div_ceil(2)) + .unwrap_or_default() +} + +fn bounded_usize_to_i32(value: usize) -> i32 { + i32::try_from(value).unwrap_or(i32::MAX) +} + +fn clamp_ask_disks_to_available(ask_disks: i32, available_disks: usize) -> i32 { + ask_disks.min(bounded_usize_to_i32(available_disks)) +} + +fn expand_ask_disks_for_object_quorum(ask_disks: i32, available_disks: usize, object_quorum: usize) -> i32 { + let Some(asked_disks) = positive_ask_disks(ask_disks) else { + return ask_disks; + }; + + bounded_usize_to_i32(asked_disks.max(object_quorum).min(available_disks)) +} + +fn write_quorum_for_drive_count(drive_count: usize, parity_count: usize) -> usize { + if drive_count == 0 { + return 0; + } + + let data_drives = drive_count.saturating_sub(parity_count); + if data_drives == parity_count { + data_drives.saturating_add(1) + } else { + data_drives + } +} + fn parse_version_marker(marker: String) -> Result { Ok(VersionMarker::parse(marker)?) } @@ -1024,28 +1574,49 @@ impl ECStore { } } - if set.set_drive_count == 4 || ask_disks > disks.len() as i32 { - ask_disks = disks.len() as i32; + if set.set_drive_count == 4 { + ask_disks = bounded_usize_to_i32(disks.len()); + } else if ask_disks > bounded_usize_to_i32(disks.len()) { + ask_disks = clamp_ask_disks_to_available(ask_disks, disks.len()); } + let listing_quorum = listing_quorum_from_ask_disks(ask_disks); + let enforce_write_quorum = enforce_latest_listing_write_quorum(opts.latest_only, &opts.ask_disks); + let write_quorum_parity = set.default_parity_count; + let required_obj_quorum = latest_listing_required_object_quorum( + listing_quorum, + set.set_drive_count, + write_quorum_parity, + enforce_write_quorum, + ); + ask_disks = expand_ask_disks_for_object_quorum(ask_disks, disks.len(), required_obj_quorum); let fallback_disks = { - if ask_disks > 0 && disks.len() > ask_disks as usize { + if let Some(asked_disks) = positive_ask_disks(ask_disks) + && disks.len() > asked_disks + { let mut rand = rand::rng(); disks.shuffle(&mut rand); - disks.split_off(ask_disks as usize) + disks.split_off(asked_disks) } else { Vec::new() } }; + let fallback_disks = Arc::new(fallback_disks); + let claim_tracker = FallbackClaimTracker::default(); - let listing_quorum = ((ask_disks + 1) / 2) as usize; + let obj_quorum = latest_listing_object_quorum( + listing_quorum, + set.set_drive_count, + write_quorum_parity, + enforce_write_quorum, + ); + let raw_min_disks = latest_listing_raw_min_disks(listing_quorum, obj_quorum, enforce_write_quorum); - let resolver = MetadataResolutionParams { - dir_quorum: listing_quorum, - obj_quorum: listing_quorum, - bucket: bucket.to_owned(), - ..Default::default() - }; + let resolver = + list_metadata_resolution_params(bucket.to_owned(), listing_quorum, obj_quorum, !opts.latest_only); + let agreed_resolver = resolver.clone(); + let partial_resolver = resolver.clone(); + let reader_disks = disks.len(); let path = base_dir_from_prefix(prefix); ensure_non_empty_listing_disks(bucket, &path, &disks)?; @@ -1066,8 +1637,23 @@ impl ECStore { let tx1 = sender.clone(); let tx2 = sender.clone(); + let supplement = ListingSupplement::new( + ListingSupplementOptions { + bucket: bucket.to_owned(), + path: path.clone(), + recursive: true, + filter_prefix: Some(filter_prefix.clone()), + forward_to: opts.marker.clone(), + per_disk_limit: bounded_usize_to_i32(opts.limit), + skip_total_timeout: false, + }, + fallback_disks.clone(), + claim_tracker.clone(), + ); + let agreed_supplement = supplement.clone(); + let partial_supplement = supplement; - list_path_raw( + list_path_raw_with_claim_tracker( rx_clone, ListPathRawOptions { disks: disks.iter().cloned().map(Some).collect(), @@ -1077,12 +1663,37 @@ impl ECStore { recursive: true, filter_prefix: Some(filter_prefix), forward_to: opts.marker.clone(), - min_disks: listing_quorum, - per_disk_limit: opts.limit as i32, + min_disks: raw_min_disks, + per_disk_limit: bounded_usize_to_i32(opts.limit), agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); + let resolver = agreed_resolver.clone(); + let supplement = agreed_supplement.clone(); async move { + let entry = match resolve_agreed_listing_entry( + entry, + reader_disks, + resolver.clone(), + enforce_write_quorum, + ) { + ListingEntryResolution::Resolved(entry) => entry, + ListingEntryResolution::NeedsSupplement(entry, fallback) => { + let Some(entry) = resolve_agreed_listing_entry_with_supplement( + entry, + reader_disks, + resolver, + enforce_write_quorum, + supplement, + ) + .await + .or(fallback) else { + return; + }; + entry + } + ListingEntryResolution::Rejected => return, + }; if entry.is_dir() { return; } @@ -1095,9 +1706,16 @@ impl ECStore { partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { Box::pin({ let value = tx2.clone(); - let resolver = resolver.clone(); + let resolver = partial_resolver.clone(); + let supplement = partial_supplement.clone(); async move { - if let Some(entry) = entries.resolve(resolver) + if let Some(entry) = resolve_listing_entries_with_supplement( + entries, + resolver, + enforce_write_quorum, + supplement, + ) + .await && let Err(err) = value.send(entry).await { error!("list_path send fail {:?}", err); @@ -1108,6 +1726,7 @@ impl ECStore { finished: None, ..Default::default() }, + claim_tracker, ) .await }); @@ -2054,25 +2673,41 @@ impl Sets { } } - if set.set_drive_count == 4 || ask_disks > disks.len() as i32 { - ask_disks = disks.len() as i32; + if set.set_drive_count == 4 { + ask_disks = bounded_usize_to_i32(disks.len()); + } else if ask_disks > bounded_usize_to_i32(disks.len()) { + ask_disks = clamp_ask_disks_to_available(ask_disks, disks.len()); } - let fallback_disks = if ask_disks > 0 && disks.len() > ask_disks as usize { + let listing_quorum = listing_quorum_from_ask_disks(ask_disks); + let enforce_write_quorum = enforce_latest_listing_write_quorum(opts.latest_only, &opts.ask_disks); + let write_quorum_parity = set.default_parity_count; + let required_obj_quorum = latest_listing_required_object_quorum( + listing_quorum, + set.set_drive_count, + write_quorum_parity, + enforce_write_quorum, + ); + ask_disks = expand_ask_disks_for_object_quorum(ask_disks, disks.len(), required_obj_quorum); + let fallback_disks = if let Some(asked_disks) = positive_ask_disks(ask_disks) + && disks.len() > asked_disks + { let mut rand = rand::rng(); disks.shuffle(&mut rand); - disks.split_off(ask_disks as usize) + disks.split_off(asked_disks) } else { Vec::new() }; + let fallback_disks = Arc::new(fallback_disks); + let claim_tracker = FallbackClaimTracker::default(); - let listing_quorum = ((ask_disks + 1) / 2) as usize; - let resolver = MetadataResolutionParams { - dir_quorum: listing_quorum, - obj_quorum: listing_quorum, - bucket: bucket.to_owned(), - ..Default::default() - }; + let obj_quorum = + latest_listing_object_quorum(listing_quorum, set.set_drive_count, write_quorum_parity, enforce_write_quorum); + let raw_min_disks = latest_listing_raw_min_disks(listing_quorum, obj_quorum, enforce_write_quorum); + let resolver = list_metadata_resolution_params(bucket.to_owned(), listing_quorum, obj_quorum, !opts.latest_only); + let agreed_resolver = resolver.clone(); + let partial_resolver = resolver.clone(); + let reader_disks = disks.len(); let path = base_dir_from_prefix(prefix); ensure_non_empty_listing_disks(bucket, &path, &disks)?; @@ -2088,8 +2723,23 @@ impl Sets { let tx1 = sender.clone(); let tx2 = sender.clone(); + let supplement = ListingSupplement::new( + ListingSupplementOptions { + bucket: bucket.to_owned(), + path: path.clone(), + recursive: true, + filter_prefix: Some(filter_prefix.clone()), + forward_to: opts.marker.clone(), + per_disk_limit: bounded_usize_to_i32(opts.limit), + skip_total_timeout: false, + }, + fallback_disks.clone(), + claim_tracker.clone(), + ); + let agreed_supplement = supplement.clone(); + let partial_supplement = supplement; - list_path_raw( + list_path_raw_with_claim_tracker( rx_clone, ListPathRawOptions { disks: disks.iter().cloned().map(Some).collect(), @@ -2099,12 +2749,37 @@ impl Sets { recursive: true, filter_prefix: Some(filter_prefix), forward_to: opts.marker.clone(), - min_disks: listing_quorum, - per_disk_limit: opts.limit as i32, + min_disks: raw_min_disks, + per_disk_limit: bounded_usize_to_i32(opts.limit), agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); + let resolver = agreed_resolver.clone(); + let supplement = agreed_supplement.clone(); async move { + let entry = match resolve_agreed_listing_entry( + entry, + reader_disks, + resolver.clone(), + enforce_write_quorum, + ) { + ListingEntryResolution::Resolved(entry) => entry, + ListingEntryResolution::NeedsSupplement(entry, fallback) => { + let Some(entry) = resolve_agreed_listing_entry_with_supplement( + entry, + reader_disks, + resolver, + enforce_write_quorum, + supplement, + ) + .await + .or(fallback) else { + return; + }; + entry + } + ListingEntryResolution::Rejected => return, + }; if entry.is_dir() { return; } @@ -2117,9 +2792,16 @@ impl Sets { partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { Box::pin({ let value = tx2.clone(); - let resolver = resolver.clone(); + let resolver = partial_resolver.clone(); + let supplement = partial_supplement.clone(); async move { - if let Some(entry) = entries.resolve(resolver) + if let Some(entry) = resolve_listing_entries_with_supplement( + entries, + resolver, + enforce_write_quorum, + supplement, + ) + .await && let Err(err) = value.send(entry).await { error!("list_path send fail {:?}", err); @@ -2130,6 +2812,7 @@ impl Sets { finished: None, ..Default::default() }, + claim_tracker, ) .await }); @@ -2802,25 +3485,46 @@ impl SetDisks { } } - if self.set_drive_count == 4 || ask_disks > disks.len() as i32 { - ask_disks = disks.len() as i32; + if self.set_drive_count == 4 { + ask_disks = bounded_usize_to_i32(disks.len()); + } else if ask_disks > bounded_usize_to_i32(disks.len()) { + ask_disks = clamp_ask_disks_to_available(ask_disks, disks.len()); } - let listing_quorum = ((ask_disks + 1) / 2) as usize; + let listing_quorum = listing_quorum_from_ask_disks(ask_disks); ensure_non_empty_listing_disks(&opts.bucket, &opts.base_dir, &disks)?; + let enforce_write_quorum = enforce_latest_listing_write_quorum(!opts.versioned, &opts.ask_disks); + let write_quorum_parity = self.default_parity_count; + let required_obj_quorum = latest_listing_required_object_quorum( + listing_quorum, + self.set_drive_count, + write_quorum_parity, + enforce_write_quorum, + ); + ask_disks = expand_ask_disks_for_object_quorum(ask_disks, disks.len(), required_obj_quorum); let mut fallback_disks = Vec::new(); - if ask_disks > 0 && disks.len() > ask_disks as usize { + if let Some(asked_disks) = positive_ask_disks(ask_disks) + && disks.len() > asked_disks + { let mut rand = rand::rng(); disks.shuffle(&mut rand); - fallback_disks = disks.split_off(ask_disks as usize); + fallback_disks = disks.split_off(asked_disks); } + let fallback_disks = Arc::new(fallback_disks); + let claim_tracker = FallbackClaimTracker::default(); let bucket = opts.bucket.clone(); let base_dir = opts.base_dir.clone(); - let resolver = list_metadata_resolution_params(bucket.clone(), listing_quorum, opts.versioned); + let latest_object_quorum = + latest_listing_object_quorum(listing_quorum, self.set_drive_count, write_quorum_parity, enforce_write_quorum); + let raw_min_disks = latest_listing_raw_min_disks(listing_quorum, latest_object_quorum, enforce_write_quorum); + let resolver = list_metadata_resolution_params(bucket.clone(), listing_quorum, latest_object_quorum, opts.versioned); + let agreed_resolver = resolver.clone(); + let partial_resolver = resolver.clone(); + let reader_disks = disks.len(); debug!( bucket = %bucket, @@ -2828,6 +3532,8 @@ impl SetDisks { set_drive_count = self.set_drive_count, asked_disks = ask_disks, listing_quorum = listing_quorum, + latest_object_quorum = latest_object_quorum, + raw_min_disks = raw_min_disks, fallback_disks = fallback_disks.len(), limit = opts.limit, stop_disk_at_limit = opts.stop_disk_at_limit, @@ -2846,8 +3552,23 @@ impl SetDisks { let tx2 = sender.clone(); let cancel_for_send1 = rx.clone(); let cancel_for_send2 = rx.clone(); + let supplement = ListingSupplement::new( + ListingSupplementOptions { + bucket: bucket.clone(), + path: opts.base_dir.clone(), + recursive: opts.recursive, + filter_prefix: opts.filter_prefix.clone(), + forward_to: opts.marker.clone(), + per_disk_limit: limit, + skip_total_timeout: false, + }, + fallback_disks.clone(), + claim_tracker.clone(), + ); + let agreed_supplement = supplement.clone(); + let partial_supplement = supplement; - let result = list_path_raw( + let result = list_path_raw_with_claim_tracker( rx, ListPathRawOptions { disks: disks.iter().cloned().map(Some).collect(), @@ -2857,13 +3578,35 @@ impl SetDisks { recursive: opts.recursive, filter_prefix: opts.filter_prefix, forward_to: opts.marker, - min_disks: listing_quorum, + min_disks: raw_min_disks, per_disk_limit: limit, agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); let cancel_token = cancel_for_send1.clone(); + let resolver = agreed_resolver.clone(); + let supplement = agreed_supplement.clone(); async move { + let entry = + match resolve_agreed_listing_entry(entry, reader_disks, resolver.clone(), enforce_write_quorum) { + ListingEntryResolution::Resolved(entry) => entry, + ListingEntryResolution::NeedsSupplement(entry, fallback) => { + let Some(entry) = resolve_agreed_listing_entry_with_supplement( + entry, + reader_disks, + resolver, + enforce_write_quorum, + supplement, + ) + .await + .or(fallback) else { + return; + }; + entry + } + ListingEntryResolution::Rejected => return, + }; + if let Err(err) = value.send(entry).await && !cancel_token.is_cancelled() { @@ -2875,10 +3618,12 @@ impl SetDisks { partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { Box::pin({ let value = tx2.clone(); - let resolver = resolver.clone(); + let resolver = partial_resolver.clone(); let cancel_token = cancel_for_send2.clone(); + let supplement = partial_supplement.clone(); async move { - if let Some(entry) = entries.resolve(resolver) + if let Some(entry) = + resolve_listing_entries_with_supplement(entries, resolver, enforce_write_quorum, supplement).await && let Err(err) = value.send(entry).await && !cancel_token.is_cancelled() { @@ -2890,6 +3635,7 @@ impl SetDisks { finished: None, ..Default::default() }, + claim_tracker, ) .await; @@ -2989,13 +3735,21 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 { #[cfg(test)] mod test { use super::{ - ENV_API_LIST_OBJECTS_QUORUM, ENV_API_LIST_QUORUM, GatherResultsState, ListPathOptions, MAX_OBJECT_LIST, VersionMarker, - gather_results, list_metadata_resolution_params, list_objects_quorum_from_env, list_quorum_from_env, max_keys_plus_one, - merge_entry_channels, normalize_list_quorum, parse_version_marker, version_marker_for_entries, + ENV_API_LIST_OBJECTS_QUORUM, ENV_API_LIST_QUORUM, FallbackListingEntries, FallbackListingEntry, GatherResultsState, + ListPathOptions, ListPathRawOptions, ListingEntryResolution, ListingSupplement, ListingSupplementOptions, + MAX_OBJECT_LIST, VersionMarker, enforce_latest_listing_write_quorum, expand_ask_disks_for_object_quorum, + fallback_entries_for_object, gather_results, latest_listing_allow_agreed_objects, latest_listing_object_quorum, + latest_listing_raw_min_disks, latest_listing_required_object_quorum, list_metadata_resolution_params, + list_objects_quorum_from_env, list_quorum_from_env, max_keys_plus_one, merge_entry_channels, normalize_list_quorum, + parse_version_marker, resolve_agreed_listing_entry, resolve_listing_entries, version_marker_for_entries, walk_result_from_set_errors, }; + use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw}; + use crate::disk::{DiskAPI, DiskOption, endpoint::Endpoint, error::DiskError, new_disk}; use crate::error::StorageError; - use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry}; + use rustfs_filemeta::{FileInfo, FileMeta, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry}; + use std::collections::{HashMap, HashSet}; + use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::mpsc; use tokio::time::timeout; @@ -3009,6 +3763,177 @@ mod test { } } + fn test_object_meta_entry(name: &str) -> MetaCacheEntry { + let mut meta = FileMeta::new(); + meta.add_version(FileInfo { + volume: "bucket".to_owned(), + name: name.to_owned(), + size: 1, + mod_time: Some(time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")), + ..Default::default() + }) + .expect("test metadata should accept object version"); + let metadata = meta.marshal_msg().expect("test metadata should marshal"); + + MetaCacheEntry { + name: name.to_owned(), + metadata, + cached: Some(meta), + reusable: false, + } + } + + #[test] + fn fallback_entries_for_object_filters_claimed_physical_disks() { + let mut entries = FallbackListingEntries::new(); + entries.insert( + "object".to_string(), + vec![ + FallbackListingEntry { + endpoint: "disk-a".to_string(), + entry: test_meta_entry("object"), + }, + FallbackListingEntry { + endpoint: "disk-b".to_string(), + entry: test_meta_entry("object"), + }, + ], + ); + let claimed = HashSet::from(["disk-a".to_string()]); + + let candidates = fallback_entries_for_object(&entries, "object", &claimed); + + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].as_ref().map(|entry| entry.name.as_str()), Some("object")); + } + + #[test] + fn listing_supplement_uses_page_cache_only_for_bounded_walks() { + let bounded = ListingSupplement::new( + ListingSupplementOptions { + bucket: "bucket".to_string(), + path: String::new(), + recursive: true, + filter_prefix: None, + forward_to: None, + per_disk_limit: 100, + skip_total_timeout: true, + }, + Arc::new(Vec::new()), + FallbackClaimTracker::default(), + ); + let unbounded = ListingSupplement::new( + ListingSupplementOptions { + bucket: "bucket".to_string(), + path: String::new(), + recursive: true, + filter_prefix: None, + forward_to: None, + per_disk_limit: 0, + skip_total_timeout: false, + }, + Arc::new(Vec::new()), + FallbackClaimTracker::default(), + ); + + assert!(bounded.entries.is_some()); + assert!(bounded.options.skip_total_timeout); + assert!(unbounded.entries.is_none()); + assert!(!unbounded.options.skip_total_timeout); + } + + #[tokio::test] + async fn listing_supplement_unbounded_reads_object_metadata_from_fallback_disk() { + let tempdir = tempfile::tempdir().expect("tempdir should be created"); + let endpoint = + Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("local disk should be created"); + let mut fi = FileInfo::new("object", 1, 1); + fi.volume = "bucket".to_owned(); + fi.name = "object".to_owned(); + fi.size = 1; + fi.fresh = true; + fi.erasure.index = 1; + fi.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")); + fi.metadata.insert("etag".to_owned(), "object-etag".to_owned()); + disk.write_metadata("", "bucket", "object", fi) + .await + .expect("test metadata should be written"); + + let supplement = ListingSupplement::new( + ListingSupplementOptions { + bucket: "bucket".to_string(), + path: String::new(), + recursive: true, + filter_prefix: None, + forward_to: None, + per_disk_limit: 0, + skip_total_timeout: false, + }, + Arc::new(vec![disk]), + FallbackClaimTracker::default(), + ); + + let entries = supplement.entries_for("object").await; + + assert!(supplement.entries.is_none()); + assert_eq!(entries.len(), 1); + let entry = entries + .into_iter() + .next() + .and_then(|entry| entry) + .expect("fallback object metadata should be returned"); + let info = entry.to_fileinfo("bucket").expect("fallback metadata should decode"); + assert_eq!(entry.name, "object"); + assert_eq!(info.metadata.get("etag").map(String::as_str), Some("object-etag")); + } + + fn test_object_meta_entry_with_erasure_versions( + name: &str, + versions: &[(time::OffsetDateTime, &str, usize, usize)], + ) -> MetaCacheEntry { + let mut meta = FileMeta::new(); + for (idx, (mod_time, etag, data_blocks, parity_blocks)) in versions.iter().enumerate() { + let mut metadata = HashMap::new(); + metadata.insert("etag".to_string(), (*etag).to_string()); + + let mut fi = FileInfo::new(name, *data_blocks, *parity_blocks); + fi.volume = "bucket".to_owned(); + fi.name = name.to_owned(); + let version_idx = u128::try_from(idx + 1).expect("test version index should fit u128"); + fi.version_id = Some(Uuid::from_u128(version_idx)); + fi.versioned = true; + fi.size = 1; + fi.mod_time = Some(*mod_time); + fi.metadata = metadata; + + meta.add_version(fi).expect("test metadata should accept object version"); + } + let metadata = meta.marshal_msg().expect("test metadata should marshal"); + + MetaCacheEntry { + name: name.to_owned(), + metadata, + cached: Some(meta), + reusable: false, + } + } + + fn test_dir_meta_entry(name: &str) -> MetaCacheEntry { + MetaCacheEntry { + name: name.to_owned(), + ..Default::default() + } + } + fn sorted_entries(names: &[&str]) -> MetaCacheEntriesSorted { MetaCacheEntriesSorted { o: MetaCacheEntries(names.iter().map(|name| Some(test_meta_entry(name))).collect()), @@ -3295,17 +4220,17 @@ mod test { #[test] fn list_metadata_resolution_params_limits_plain_listing_to_latest_version() { - let resolver = list_metadata_resolution_params("bucket".to_string(), 2, false); + let resolver = list_metadata_resolution_params("bucket".to_string(), 2, 3, false); - assert_eq!(resolver.dir_quorum, 2); - assert_eq!(resolver.obj_quorum, 2); + assert_eq!(resolver.dir_quorum, 3); + assert_eq!(resolver.obj_quorum, 3); assert_eq!(resolver.bucket, "bucket"); assert_eq!(resolver.requested_versions, 1); } #[test] fn list_metadata_resolution_params_keeps_all_versions_for_version_listing() { - let resolver = list_metadata_resolution_params("bucket".to_string(), 3, true); + let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 5, true); assert_eq!(resolver.dir_quorum, 3); assert_eq!(resolver.obj_quorum, 3); @@ -3313,6 +4238,287 @@ mod test { assert_eq!(resolver.requested_versions, 0); } + #[test] + fn latest_listing_object_quorum_uses_write_quorum_for_strict_latest_listing() { + let required_quorum = latest_listing_required_object_quorum(2, 8, 4, true); + let ask_disks = expand_ask_disks_for_object_quorum(4, 8, required_quorum); + + assert_eq!(required_quorum, 5); + assert_eq!(ask_disks, 5); + assert_eq!(latest_listing_object_quorum(2, 8, 4, true), 5); + } + + #[test] + fn latest_listing_object_quorum_calculates_low_parity_write_quorum() { + let required_quorum = latest_listing_required_object_quorum(2, 8, 1, true); + let ask_disks = expand_ask_disks_for_object_quorum(4, 8, required_quorum); + + assert_eq!(required_quorum, 7); + assert_eq!(ask_disks, 7); + assert_eq!(latest_listing_object_quorum(2, 8, 1, true), 7); + } + + #[test] + fn latest_listing_object_quorum_keeps_reduced_and_versioned_listing_quorum() { + assert!(!enforce_latest_listing_write_quorum(true, "reduced")); + assert!(!enforce_latest_listing_write_quorum(true, "disk")); + assert!(enforce_latest_listing_write_quorum(true, "optimal")); + assert!(!enforce_latest_listing_write_quorum(false, "optimal")); + assert_eq!(latest_listing_required_object_quorum(1, 4, 2, false), 1); + assert_eq!(latest_listing_object_quorum(1, 4, 2, false), 1); + assert_eq!(latest_listing_required_object_quorum(2, 4, 2, false), 2); + assert_eq!(latest_listing_object_quorum(2, 4, 2, false), 2); + assert_eq!(expand_ask_disks_for_object_quorum(2, 4, 2), 2); + } + + #[test] + fn latest_listing_object_quorum_requires_write_quorum_when_degraded_cannot_satisfy_it() { + let required_quorum = latest_listing_required_object_quorum(2, 8, 4, true); + let ask_disks = expand_ask_disks_for_object_quorum(4, 4, required_quorum); + + assert_eq!(required_quorum, 5); + assert_eq!(ask_disks, 4); + assert_eq!(latest_listing_object_quorum(2, 8, 4, true), 5); + } + + #[tokio::test] + async fn latest_listing_agreed_path_requires_enough_readers_for_write_quorum() { + let entry = test_object_meta_entry("object"); + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + let allow_agreed_objects = latest_listing_allow_agreed_objects(true, 4, 5); + let raw_min_disks = latest_listing_raw_min_disks(2, 5, true); + + assert!(!allow_agreed_objects); + assert_eq!(raw_min_disks, 5); + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None], + min_disks: raw_min_disks, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry.clone()]); 4], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + Box::pin(async move { + if !allow_agreed_objects && !entry.is_dir() { + return; + } + seen.lock().expect("seen mutex poisoned").push(entry.name); + }) + })), + ..Default::default() + }, + ) + .await + .expect_err("latest listing should fail when selected readers cannot satisfy write quorum"); + + assert_eq!(err, DiskError::ErasureReadQuorum); + assert!(seen.lock().expect("seen mutex poisoned").is_empty()); + + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + let allow_agreed_objects = latest_listing_allow_agreed_objects(true, 5, 5); + let raw_min_disks = latest_listing_raw_min_disks(3, 5, true); + + assert!(allow_agreed_objects); + assert_eq!(raw_min_disks, 5); + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None, None], + min_disks: raw_min_disks, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry]); 5], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + Box::pin(async move { + if !allow_agreed_objects && !entry.is_dir() { + return; + } + seen.lock().expect("seen mutex poisoned").push(entry.name); + }) + })), + ..Default::default() + }, + ) + .await + .expect("write-quorum all-agree listing should complete"); + + assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["object".to_string()]); + } + + #[tokio::test] + async fn latest_listing_agreed_path_reconciles_low_parity_partial_latest() { + let old_mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let new_mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let entry = test_object_meta_entry_with_erasure_versions( + "object", + &[(old_mod_time, "old-etag", 4, 4), (new_mod_time, "new-etag", 7, 1)], + ); + let fallback_old_entry = test_object_meta_entry_with_erasure_versions("object", &[(old_mod_time, "old-etag", 4, 4)]); + let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 5, false); + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None, None], + min_disks: 5, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry]); 5], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + let resolver = resolver.clone(); + let fallback_old_entry = fallback_old_entry.clone(); + Box::pin(async move { + let entry = match resolve_agreed_listing_entry(entry, 5, resolver.clone(), true) { + ListingEntryResolution::Resolved(entry) => entry, + ListingEntryResolution::NeedsSupplement(entry, _) => { + let mut candidates = vec![Some(entry); 5]; + candidates.extend(vec![Some(fallback_old_entry); 2]); + let Some(entry) = resolve_listing_entries(MetaCacheEntries(candidates), resolver, true) else { + return; + }; + entry + } + ListingEntryResolution::Rejected => return, + }; + let info = entry.to_fileinfo("bucket").expect("resolved entry should decode"); + seen.lock().expect("seen mutex poisoned").push(( + entry.name, + info.mod_time, + info.metadata.get("etag").cloned(), + )); + }) + })), + ..Default::default() + }, + ) + .await + .expect("all-agree listing should complete"); + + assert_eq!( + seen.lock().expect("seen mutex poisoned").as_slice(), + &[("object".to_string(), Some(old_mod_time), Some("old-etag".to_string()))] + ); + } + + #[tokio::test] + async fn latest_listing_agreed_path_falls_back_to_previous_version_without_supplement() { + let old_mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let new_mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let entry = test_object_meta_entry_with_erasure_versions( + "object", + &[(old_mod_time, "old-etag", 4, 4), (new_mod_time, "new-etag", 7, 1)], + ); + let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 5, false); + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None, None], + min_disks: 5, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry]); 5], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + let resolver = resolver.clone(); + Box::pin(async move { + let entry = match resolve_agreed_listing_entry(entry, 5, resolver, true) { + ListingEntryResolution::Resolved(entry) => entry, + ListingEntryResolution::NeedsSupplement(_, Some(entry)) => entry, + ListingEntryResolution::NeedsSupplement(_, None) | ListingEntryResolution::Rejected => return, + }; + let info = entry.to_fileinfo("bucket").expect("resolved entry should decode"); + seen.lock().expect("seen mutex poisoned").push(( + entry.name, + info.mod_time, + info.metadata.get("etag").cloned(), + )); + }) + })), + ..Default::default() + }, + ) + .await + .expect("all-agree listing should complete"); + + assert_eq!( + seen.lock().expect("seen mutex poisoned").as_slice(), + &[("object".to_string(), Some(old_mod_time), Some("old-etag".to_string()))] + ); + } + + #[tokio::test] + async fn latest_listing_agreed_path_supplements_committed_low_parity_latest() { + let new_mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let entry = test_object_meta_entry_with_erasure_versions("object", &[(new_mod_time, "new-etag", 7, 1)]); + let fallback_entry = entry.clone(); + let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 5, false); + let seen = Arc::new(Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None, None], + min_disks: 5, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry]); 5], + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let seen = seen_clone.clone(); + let resolver = resolver.clone(); + let fallback_entry = fallback_entry.clone(); + Box::pin(async move { + let ListingEntryResolution::NeedsSupplement(entry, _) = + resolve_agreed_listing_entry(entry, 5, resolver.clone(), true) + else { + return; + }; + let mut candidates = vec![Some(entry); 5]; + candidates.extend(vec![Some(fallback_entry); 2]); + let resolved = resolve_listing_entries(MetaCacheEntries(candidates), resolver, true) + .expect("supplemented low-parity latest should satisfy object write quorum"); + let info = resolved.to_fileinfo("bucket").expect("resolved entry should decode"); + seen.lock().expect("seen mutex poisoned").push(( + resolved.name, + info.mod_time, + info.metadata.get("etag").cloned(), + )); + }) + })), + ..Default::default() + }, + ) + .await + .expect("all-agree listing should complete"); + + assert_eq!( + seen.lock().expect("seen mutex poisoned").as_slice(), + &[("object".to_string(), Some(new_mod_time), Some("new-etag".to_string()))] + ); + } + + #[tokio::test] + async fn latest_listing_agreed_directory_requires_write_quorum() { + let entry = test_dir_meta_entry("prefix/"); + let raw_min_disks = latest_listing_raw_min_disks(2, 5, true); + + let err = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None, None], + min_disks: raw_min_disks, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![entry]); 4], + agreed: Some(Box::new(move |_| Box::pin(async {}))), + ..Default::default() + }, + ) + .await + .expect_err("latest listing should not expose prefixes below write quorum"); + + assert_eq!(err, DiskError::ErasureReadQuorum); + } + #[test] fn test_null_version_marker_handling() { let parsed = parse_version_marker("null".to_string()).expect("null marker should parse"); diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index 272ec6c66..3bfe5c092 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -271,6 +271,18 @@ pub struct FileMetaShallowVersion { pub meta: Vec, // FileMetaVersion.marshal_msg } +fn write_quorum_from_erasure(data_blocks: usize, parity_blocks: usize) -> Option { + if data_blocks == 0 { + return None; + } + + Some(if data_blocks == parity_blocks { + data_blocks.saturating_add(1) + } else { + data_blocks + }) +} + impl FileMetaShallowVersion { /// Parse version meta with legacy format compatibility. /// Use this instead of `FileMetaVersion::default()` + `unmarshal_msg()` to handle old-version xl.meta. @@ -278,6 +290,18 @@ impl FileMetaShallowVersion { FileMetaVersion::try_from(self.meta.as_slice()) } + pub fn write_quorum(&self, fallback_quorum: usize) -> usize { + let header_quorum = self.header.write_quorum(fallback_quorum); + if self.header.has_ec() || !matches!(self.header.version_type, VersionType::Object | VersionType::Legacy) { + return header_quorum; + } + + self.parse_version_meta() + .ok() + .and_then(|version| version.erasure_write_quorum()) + .unwrap_or(header_quorum) + } + pub fn into_fileinfo(&self, volume: &str, path: &str, all_parts: bool) -> Result { let file_version = self.parse_version_meta()?; Ok(file_version.into_fileinfo(volume, path, all_parts)) @@ -312,6 +336,23 @@ pub struct FileMetaVersion { } impl FileMetaVersion { + fn erasure_write_quorum(&self) -> Option { + let (data_blocks, parity_blocks) = match self.version_type { + VersionType::Object | VersionType::Legacy => { + if let Some(object) = &self.object { + (object.erasure_m, object.erasure_n) + } else if let Some(object) = &self.legacy_object { + (object.erasure.data_blocks, object.erasure.parity_blocks) + } else { + return None; + } + } + _ => return None, + }; + + write_quorum_from_erasure(data_blocks, parity_blocks) + } + fn decode_data_dir_from_v2_object(buf: &[u8]) -> Result> { let mut cur = std::io::Cursor::new(buf); let mut fields = rmp::decode::read_map_len(&mut cur)?; @@ -775,6 +816,14 @@ impl FileMetaVersionHeader { self.ec_m > 0 && self.ec_n > 0 } + pub fn write_quorum(&self, fallback_quorum: usize) -> usize { + if self.version_type != VersionType::Object || !self.has_ec() { + return fallback_quorum; + } + + write_quorum_from_erasure(usize::from(self.ec_m), usize::from(self.ec_n)).unwrap_or(fallback_quorum) + } + pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool { let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o); if self.version_id.is_none() { @@ -2543,9 +2592,28 @@ pub enum Flags { // mergeXLV2Versions pub fn merge_file_meta_versions( + quorum: usize, + strict: bool, + requested_versions: usize, + versions: &[Vec], +) -> Vec { + merge_file_meta_versions_inner(quorum, strict, requested_versions, false, versions) +} + +pub(crate) fn merge_file_meta_versions_with_write_quorum( + quorum: usize, + strict: bool, + requested_versions: usize, + versions: &[Vec], +) -> Vec { + merge_file_meta_versions_inner(quorum, strict, requested_versions, true, versions) +} + +fn merge_file_meta_versions_inner( mut quorum: usize, mut strict: bool, requested_versions: usize, + enforce_write_quorum: bool, versions: &[Vec], ) -> Vec { if quorum == 0 { @@ -2557,13 +2625,32 @@ pub fn merge_file_meta_versions( } if versions.len() == 1 { - return versions[0].clone(); + if !enforce_write_quorum { + return versions[0].clone(); + } + + let required_quorum = versions[0] + .first() + .map(|version| version.write_quorum(quorum).max(quorum)) + .unwrap_or(quorum); + if versions.len() >= required_quorum { + return versions[0].clone(); + } + return Vec::new(); } if quorum == 1 { strict = true; } + let required_quorum = |version: &FileMetaShallowVersion| { + if enforce_write_quorum { + version.write_quorum(quorum).max(quorum) + } else { + quorum + } + }; + let mut versions = versions.to_owned(); let mut n_versions = 0; @@ -2594,9 +2681,12 @@ pub fn merge_file_meta_versions( let mut latest = FileMetaShallowVersion::default(); if consistent { - merged.push(tops[0].clone()); - if !tops[0].header.free_version() { - n_versions += 1; + latest = tops[0].clone(); + if tops.len() >= required_quorum(&latest) { + merged.push(latest.clone()); + if !latest.header.free_version() { + n_versions += 1; + } } } else { let mut latest_count = 0; @@ -2659,7 +2749,7 @@ pub fn merge_file_meta_versions( break; } } - if latest_count >= quorum { + if latest_count >= required_quorum(&latest) { if !latest.header.free_version() { n_versions += 1; } @@ -3053,6 +3143,39 @@ mod tests { wr } + fn encode_legacy_v2_object_body(data_blocks: usize, parity_blocks: usize) -> Vec { + let drive_count = data_blocks + parity_blocks; + let payload = LegacyObjectVersionFixture { + version_type: LegacyObjectVersionTypeFixture::Object, + object: Some(LegacyObjectFixture { + version_id: Some(sample_version_id().as_bytes().to_vec()), + data_dir: Some(Uuid::from_u128(42).as_bytes().to_vec()), + erasure_algorithm: "ReedSolomon".to_string(), + erasure_m: data_blocks, + erasure_n: parity_blocks, + erasure_block_size: 1_048_576, + erasure_index: 1, + erasure_dist: (1..=drive_count) + .map(|idx| u8::try_from(idx).expect("test drive index should fit u8")) + .collect(), + bitrot_checksum_algo: "HighwayHash".to_string(), + part_numbers: vec![1], + part_etags: vec!["etag-1".to_string()], + part_sizes: vec![11], + part_actual_sizes: vec![11], + part_indices: vec![Vec::new()], + size: 11, + mod_time: Some(sample_mod_time()), + meta_sys: HashMap::new(), + meta_user: HashMap::from([("content-type".to_string(), "text/plain".to_string())]), + }), + delete_marker: None, + write_version: 3, + }; + + rmp_serde::to_vec_named(&payload).expect("legacy object payload should marshal") + } + #[test] fn version_header_unmarshal_v1_uses_legacy_layout_defaults() { let expected = sample_header(); @@ -3087,6 +3210,38 @@ mod tests { assert_eq!(decoded.ec_m, 0); } + #[test] + fn shallow_version_write_quorum_uses_legacy_object_payload_when_header_lacks_ec() { + let expected = sample_header(); + let encoded = encode_v2_header(&expected); + let mut header = FileMetaVersionHeader::default(); + header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode"); + + let version = FileMetaShallowVersion { + header, + meta: encode_legacy_v2_object_body(7, 1), + }; + + assert_eq!(version.header.write_quorum(5), 5); + assert_eq!(version.write_quorum(5), 7); + } + + #[test] + fn shallow_version_write_quorum_uses_zero_parity_object_payload() { + let expected = sample_header(); + let encoded = encode_v2_header(&expected); + let mut header = FileMetaVersionHeader::default(); + header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode"); + + let version = FileMetaShallowVersion { + header, + meta: encode_legacy_v2_object_body(4, 0), + }; + + assert_eq!(version.header.write_quorum(2), 2); + assert_eq!(version.write_quorum(2), 4); + } + #[test] fn version_header_unmarshal_v3_round_trips_current_layout() { let expected = sample_header(); diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 42234f9b1..b44a14254 100644 --- a/crates/filemeta/src/metacache.rs +++ b/crates/filemeta/src/metacache.rs @@ -14,7 +14,7 @@ use crate::{ Error, FileInfo, FileInfoOpts, FileInfoVersions, FileMeta, FileMetaShallowVersion, Result, VersionType, get_file_info, - merge_file_meta_versions, + merge_file_meta_versions, merge_file_meta_versions_with_write_quorum, }; use arc_swap::ArcSwapOption; use rmp::Marker; @@ -334,7 +334,15 @@ impl MetaCacheEntries { &self.0 } - pub fn resolve(&self, mut params: MetadataResolutionParams) -> Option { + pub fn resolve(&self, params: MetadataResolutionParams) -> Option { + self.resolve_inner(params, false) + } + + pub fn resolve_with_write_quorum(&self, params: MetadataResolutionParams) -> Option { + self.resolve_inner(params, true) + } + + fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option { if self.0.is_empty() { debug!( bucket = %params.bucket, @@ -412,7 +420,7 @@ impl MetaCacheEntries { return Some(selected); } - // If we would never be able to reach read quorum. + // If we would never be able to reach the required object quorum. if objs_valid < params.obj_quorum { debug!( objs_valid, @@ -423,13 +431,26 @@ impl MetaCacheEntries { } if objs_agree == objs_valid { - debug!( - selected = %selected.name, - objs_agree, - objs_valid, - "metacache resolve reused selected candidate because all valid object entries agreed" - ); - return Some(selected); + let required_quorum = if enforce_write_quorum { + selected + .cached + .as_ref() + .and_then(|cached| cached.versions.first()) + .map(|version| version.write_quorum(params.obj_quorum).max(params.obj_quorum)) + .unwrap_or(params.obj_quorum) + } else { + params.obj_quorum + }; + + if objs_agree >= required_quorum { + debug!( + selected = %selected.name, + objs_agree, + objs_valid, + "metacache resolve reused selected candidate because all valid object entries agreed" + ); + return Some(selected); + } } let Some(cached) = selected.cached else { @@ -437,7 +458,16 @@ impl MetaCacheEntries { return None; }; - let versions = merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, ¶ms.candidates); + let versions = if enforce_write_quorum { + merge_file_meta_versions_with_write_quorum( + params.obj_quorum, + params.strict, + params.requested_versions, + ¶ms.candidates, + ) + } else { + merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, ¶ms.candidates) + }; if versions.is_empty() { debug!( selected = %selected.name, @@ -1272,6 +1302,300 @@ mod tests { assert_ne!(extended_versions, cached.versions.len()); } + fn metacache_entry_with_mod_time(mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry { + let mut metadata = HashMap::new(); + metadata.insert("etag".to_string(), etag.to_string()); + + let mut meta = FileMeta::new(); + meta.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + size: 1, + mod_time: Some(mod_time), + metadata, + ..Default::default() + }) + .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, + } + } + + fn metacache_entry_with_erasure( + mod_time: OffsetDateTime, + etag: &str, + data_blocks: usize, + parity_blocks: usize, + ) -> MetaCacheEntry { + let mut metadata = HashMap::new(); + metadata.insert("etag".to_string(), etag.to_string()); + + let mut fi = FileInfo::new("object", data_blocks, parity_blocks); + fi.volume = "bucket".to_string(); + fi.name = "object".to_string(); + 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, + } + } + + fn metacache_entry_with_erasure_versions(versions: &[(OffsetDateTime, &str, usize, usize)]) -> MetaCacheEntry { + let mut meta = FileMeta::new(); + for (idx, (mod_time, etag, data_blocks, parity_blocks)) in versions.iter().enumerate() { + let mut metadata = HashMap::new(); + metadata.insert("etag".to_string(), (*etag).to_string()); + + let mut fi = FileInfo::new("object", *data_blocks, *parity_blocks); + fi.volume = "bucket".to_string(); + fi.name = "object".to_string(); + let version_idx = u128::try_from(idx + 1).expect("test version index should fit u128"); + fi.version_id = Some(Uuid::from_u128(version_idx)); + fi.versioned = true; + fi.size = 1; + fi.mod_time = Some(*mod_time); + fi.metadata = metadata; + + 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, + } + } + + fn metacache_entry_without_header_ec(mut entry: MetaCacheEntry) -> MetaCacheEntry { + let mut cached = entry.cached.take().expect("test entry should have cached metadata"); + for version in cached.versions.iter_mut() { + version.header.ec_m = 0; + version.header.ec_n = 0; + } + entry.metadata = cached.marshal_msg().expect("test file metadata should marshal"); + entry.cached = Some(cached); + entry + } + + fn metacache_dir_entry(name: &str) -> MetaCacheEntry { + MetaCacheEntry { + name: name.to_string(), + ..Default::default() + } + } + + #[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"); + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let old_entry = metacache_entry_with_mod_time(old_mod_time, "old-etag"); + let new_entry = metacache_entry_with_mod_time(new_mod_time, "new-etag"); + + let resolved = MetaCacheEntries(vec![ + Some(new_entry.clone()), + Some(new_entry), + Some(old_entry.clone()), + Some(old_entry.clone()), + Some(old_entry.clone()), + Some(old_entry.clone()), + Some(old_entry), + ]) + .resolve(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }) + .expect("previous committed metadata should still satisfy write quorum"); + + let info = resolved + .to_fileinfo("bucket") + .expect("resolved committed metadata should decode as file info"); + assert_eq!(info.mod_time, Some(old_mod_time)); + assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag")); + } + + #[test] + fn resolve_rejects_low_parity_partial_latest_below_required_object_quorum() { + let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 7, 1); + let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1); + + let resolved = MetaCacheEntries(vec![ + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry), + Some(old_entry.clone()), + Some(old_entry), + ]) + .resolve_with_write_quorum(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }); + + assert!(resolved.is_none()); + } + + #[test] + fn resolve_rejects_zero_parity_partial_latest_below_required_object_quorum() { + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 0); + + let resolved = MetaCacheEntries(vec![ + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry), + ]) + .resolve_with_write_quorum(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }); + + assert!(resolved.is_none()); + } + + #[test] + fn resolve_accepts_low_parity_latest_at_required_object_quorum() { + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let new_entry = metacache_entry_with_erasure(new_mod_time, "new-etag", 7, 1); + + let resolved = MetaCacheEntries(vec![ + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry.clone()), + Some(new_entry), + ]) + .resolve_with_write_quorum(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }) + .expect("latest metadata should resolve after satisfying its own write quorum"); + + let info = resolved + .to_fileinfo("bucket") + .expect("resolved committed metadata should decode as file info"); + assert_eq!(info.mod_time, Some(new_mod_time)); + assert_eq!(info.metadata.get("etag").map(String::as_str), Some("new-etag")); + } + + #[test] + fn resolve_skips_low_parity_partial_latest_and_returns_committed_previous_version() { + let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4); + let new_and_old_entry = + metacache_entry_with_erasure_versions(&[(old_mod_time, "old-etag", 4, 4), (new_mod_time, "new-etag", 7, 1)]); + + let resolved = MetaCacheEntries(vec![ + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry), + Some(old_entry.clone()), + Some(old_entry), + ]) + .resolve_with_write_quorum(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }) + .expect("previous committed metadata should resolve after rejecting partial latest"); + + let info = resolved + .to_fileinfo("bucket") + .expect("resolved committed metadata should decode as file info"); + assert_eq!(info.mod_time, Some(old_mod_time)); + assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag")); + } + + #[test] + fn resolve_skips_legacy_header_low_parity_partial_latest_using_payload_quorum() { + let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let new_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_400).expect("valid timestamp"); + let old_entry = metacache_entry_with_erasure(old_mod_time, "old-etag", 4, 4); + let new_and_old_entry = metacache_entry_without_header_ec(metacache_entry_with_erasure_versions(&[ + (old_mod_time, "old-etag", 4, 4), + (new_mod_time, "new-etag", 7, 1), + ])); + + let resolved = MetaCacheEntries(vec![ + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry.clone()), + Some(new_and_old_entry), + Some(old_entry.clone()), + Some(old_entry), + ]) + .resolve_with_write_quorum(MetadataResolutionParams { + obj_quorum: 5, + requested_versions: 1, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }) + .expect("previous committed metadata should resolve after rejecting legacy-header partial latest"); + + let info = resolved + .to_fileinfo("bucket") + .expect("resolved committed metadata should decode as file info"); + assert_eq!(info.mod_time, Some(old_mod_time)); + assert_eq!(info.metadata.get("etag").map(String::as_str), Some("old-etag")); + } + + #[test] + fn resolve_rejects_partial_directory_below_dir_quorum() { + let partial_dir = metacache_dir_entry("prefix/"); + + let resolved = MetaCacheEntries(vec![Some(partial_dir.clone()), Some(partial_dir)]).resolve(MetadataResolutionParams { + dir_quorum: 5, + obj_quorum: 5, + bucket: "bucket".to_string(), + strict: true, + ..Default::default() + }); + + assert!(resolved.is_none()); + } + fn build_hashmap_cache(update_size: usize) -> Arc>> { let generation = Arc::new(AtomicUsize::new(0)); Arc::new(Cache::new(