diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index e256dad13..01614136f 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -27,6 +27,13 @@ use rustfs_filemeta::metadata_keys; use rustfs_utils::http; use sha2::Digest; +#[derive(Clone, Copy)] +struct FileInfoIdentityGroup { + hash: [u8; 32], + count: usize, + mod_time: Option, +} + impl SetDisks { /// Read one key's complete version stack without enumerating its parent. pub(crate) async fn read_object_versions_for_lifecycle( @@ -583,7 +590,135 @@ impl SetDisks { (new_disk, mod_time, None) } - /// Select the version a read serves with MinIO's rule (`getObjectFileInfo` + 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 file_info_is_valid_for_metadata(meta) { + 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; + } + + if let Some((first, _)) = parts_metadata + .iter() + .zip(errs.iter()) + .find(|(meta, err)| err.is_none() && file_info_is_valid_for_metadata(meta)) + { + let same_count = parts_metadata + .iter() + .zip(errs.iter()) + .filter(|(meta, err)| { + err.is_none() + && file_info_is_valid_for_metadata(meta) + && super::core::metadata_quorum::metadata_early_stop_candidate_matches(first, meta) + }) + .count(); + if same_count >= write_quorum { + 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() || !file_info_is_valid_for_metadata(meta) { + 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 + } + + /// The commit-proof rule kept for internal metadata in the system bucket. + /// Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an + /// ordinary read as proof that a record committed: when enough copies + /// answer to prove write quorum, the latest version must hold it, and a + /// degraded read never serves a newer version below write quorum. A split + /// write therefore stays unreadable until native recovery resolves it. + fn select_committed_fileinfo( + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + version_id: &str, + read_quorum: usize, + write_quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo)> { + 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 + { + return Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_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)) + } + + /// Select the version a read of `bucket` serves: S3 buckets use MinIO's + /// read-quorum rule (`select_valid_fileinfo`); the system bucket keeps the + /// commit-proof rule (`select_committed_fileinfo`). + #[allow(clippy::too_many_arguments)] + pub(super) fn select_fileinfo_for_read( + bucket: &str, + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + version_id: &str, + read_quorum: usize, + write_quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo)> { + if crate::bucket::utils::is_meta_bucketname(bucket) { + Self::select_committed_fileinfo(disks, parts_metadata, errs, version_id, read_quorum, write_quorum) + } else { + Self::select_valid_fileinfo(disks, parts_metadata, errs, read_quorum) + } + } + + /// Select the version an S3 read serves with MinIO's rule (`getObjectFileInfo` /// -> `listOnlineDisks` -> `pickValidFileInfo`): the modification time /// shared by the most copies wins when it reaches `read_quorum`, a tie /// goes to the newer time, and one content identity at that time must @@ -853,6 +988,100 @@ impl SetDisks { } } + 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() || !file_info_is_valid_for_metadata(meta) { + 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() || !file_info_is_valid_for_metadata(meta) || 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, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 007c53338..963275203 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -12153,6 +12153,47 @@ mod tests { } } + #[test] + fn test_system_bucket_reads_keep_the_commit_proof_rule() { + 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 old = |index| quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", index); + let new = |index| quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", index); + + // A PREPARE/COMMIT split of pool metadata must stay unreadable so that + // native recovery, not an ordinary read, decides whether it committed. + for (metas, errs) in [ + (vec![old(1), old(2), new(3), new(4)], vec![None, None, None, None]), + ( + vec![new(1), new(2), old(3), FileInfo::default()], + vec![None, None, None, Some(DiskError::DiskNotFound)], + ), + ] { + let disks = vec![None; metas.len()]; + let (_, served) = SetDisks::select_fileinfo_for_read("bucket", &disks, &metas, &errs, "", 2, 3) + .expect("an S3 read serves the newer side at read quorum"); + assert_eq!(served.data_dir, Some(new_data_dir)); + + let internal = SetDisks::select_fileinfo_for_read(crate::disk::RUSTFS_META_BUCKET, &disks, &metas, &errs, "", 2, 3); + assert!(matches!(internal, Err(DiskError::ErasureReadQuorum))); + } + + let committed = vec![new(1), new(2), new(3), old(4)]; + let (_, internal) = SetDisks::select_fileinfo_for_read( + crate::disk::RUSTFS_META_BUCKET, + &vec![None; committed.len()], + &committed, + &[None, None, None, None], + "", + 2, + 3, + ) + .expect("a write-quorum version stays readable in the system bucket"); + assert_eq!(internal.data_dir, Some(new_data_dir)); + } + #[test] fn test_latest_fileinfo_selection_picks_version_stack_shape_in_disk_order() { let mod_time = OffsetDateTime::now_utc(); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 962927879..baede67c7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -9319,8 +9319,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } }; - let (online_disks, mut fi) = Self::select_valid_fileinfo(&disks, &metas, &errs, 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_fileinfo_for_read(src_bucket, &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() { @@ -11145,8 +11147,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((read_quorum, _)) => read_quorum, + 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) => { // A replication terminal CAS is a validation/write-back // attempt, never a heal operation. In particular, @@ -11176,9 +11178,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { 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, mut fi) = Self::select_valid_fileinfo(&disks, &metas, &errs, read_quorum) - .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; + let version_id = opts.version_id.as_deref().unwrap_or_default(); + let (online_disks, mut fi) = + Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum) + .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; if fi.deleted { return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object])); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 11d59e616..e99512a10 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -926,7 +926,7 @@ impl SetDisks { None } }; - 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| dangling_not_found(&err).unwrap_or(err)) .map_err(|err| to_object_err(err.into(), vec![bucket, object, &vid])) { @@ -944,6 +944,8 @@ impl SetDisks { }; 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]))?; if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { if let Some(not_found) = dangling_not_found(&err) { rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "not_found"); @@ -964,14 +966,19 @@ impl SetDisks { return Err(to_object_err(err.into(), vec![bucket, object, &vid])); } - let (op_online_disks, mut fi) = match Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, read_quorum) { - Ok(selection) => selection, - Err(err) => { - let err = StorageError::from(err); - rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, metadata_quorum_outcome(&err)); - return Err(err); - } - }; + let (op_online_disks, mut fi) = + match Self::select_fileinfo_for_read(bucket, &disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum) + { + Ok(selection) => selection, + Err(err) => { + let err = StorageError::from(err); + rustfs_io_metrics::record_get_object_metadata_quorum_result( + metadata_metrics_path, + metadata_quorum_outcome(&err), + ); + return Err(err); + } + }; let include_part_checksums = opts.include_part_checksums || opts.part_number.is_some() || opts.data_movement || opts.raw_data_movement_read; if include_part_checksums { @@ -2448,15 +2455,18 @@ impl SetDisks { let (mut parts_metadata, errs, diagnostics) = metadata_read.into_legacy(); diagnostics.record(metrics_path); - let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks) + let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks) .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?; 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]))?; if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) { return Err(to_object_err(err.into(), vec![bucket, object])); } - let (mut online_disks, full_fi) = SetDisks::select_valid_fileinfo(fallback_disks, &parts_metadata, &errs, read_quorum)?; + let (mut online_disks, full_fi) = + SetDisks::select_fileinfo_for_read(bucket, fallback_disks, &parts_metadata, &errs, "", read_quorum, write_quorum)?; if !late_metadata_read_identity_matches(expected, &full_fi) { return Err(to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object])); }