From 268a5519f1a0d3159ef440d6f2814333aa75cfc6 Mon Sep 17 00:00:00 2001 From: Chris An Date: Wed, 28 Oct 2026 00:07:05 +0800 Subject: [PATCH 1/2] test(ecstore): stabilize metadata fanout lifecycle checks (#8999) --- .../src/set_disk/core/io_primitives.rs | 33 +++++++++++++------ 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 118d66b9c..99178c7de 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -9440,7 +9440,7 @@ mod tests { "lifecycle-inline-data-get-fanout-bucket", "lifecycle-inline-data-get-object", GET_OBJECT_PATH_LEGACY_DUPLEX, - None, + false, ); } @@ -9451,7 +9451,7 @@ mod tests { "lifecycle-bounded-inline-data-get-fanout-bucket", "lifecycle-bounded-inline-data-get-object", GET_OBJECT_PATH_LEGACY_DUPLEX, - Some("true"), + true, ); } @@ -9462,7 +9462,7 @@ mod tests { RUSTFS_META_BUCKET, "buckets/.usage-cache/lifecycle-inline-data-get-object", GET_OBJECT_PATH_INTERNAL_META, - None, + false, ); } @@ -9563,7 +9563,7 @@ mod tests { bucket: &'static str, object_prefix: &str, expected_path: &'static str, - bounded_fanout_env: Option<&'static str>, + bounded_fanout: bool, ) { const DISKS: usize = 4; let object = object_with_initial_data_shards(bucket, object_prefix, 2, 3); @@ -9583,16 +9583,26 @@ mod tests { [ ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")), - ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", bounded_fanout_env), + ( + "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", + Some(if bounded_fanout { "true" } else { "false" }), + ), ], async { - let barrier_disk = if bounded_fanout_env.is_some() { - bounded_spare_disk_index(bucket, &object, 2, 2) - } else { - 3 - }; + let barrier_disk = bounded_spare_disk_index(bucket, &object, 2, 2); let barrier = rename_fanout_barrier::arm(&object, barrier_disk, rename_fanout_barrier::PHASE_READ_VERSION); + // Keep one initial bounded read pending so its hedge is scheduled; + // eager fanout needs all quorum reads parked until the spare starts. + let quorum_disks = if bounded_fanout { + vec![bounded_metadata_fanout_order(bucket, &object, DISKS, 2)[0]] + } else { + (0..DISKS).filter(|&index| index != barrier_disk).collect() + }; + let quorum_barriers = quorum_disks + .into_iter() + .map(|index| rename_fanout_barrier::arm(&object, index, rename_fanout_barrier::PHASE_READ_VERSION)) + .collect::>(); let disks_for_read = disks.clone(); let object_for_read = object.clone(); let mut read = tokio::spawn(async move { @@ -9614,6 +9624,9 @@ mod tests { tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused()) .await .expect("metadata fanout should pause the spare metadata task"); + for quorum_barrier in &quorum_barriers { + quorum_barrier.release(); + } let (parts_metadata, errs, diagnostics) = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read) .await .expect("verified inline quorum should abort the paused eager fanout") From acebee4ea349f3422cc3a8ef8147867f78ea077e Mon Sep 17 00:00:00 2001 From: Chris An Date: Wed, 28 Oct 2026 00:47:50 +0800 Subject: [PATCH 2/2] fix(ecstore): select the latest version at read quorum like MinIO (#8982) * fix(ecstore): select the latest version at read quorum like MinIO GET and HEAD escalated latest-version selection to write quorum when enough metadata answered and skipped a newer version below write quorum on degraded reads. LIST and heal already pick the newer side of a split write at read quorum, so such keys listed but answered 503 until heal. Select with MinIO's getObjectFileInfo rule: the most common mod time that reaches the object's data blocks wins, ties go to the newer time, and one content identity at that time must reach the same quorum. Two identities tied at the selected time fail instead of resolving by hash order, and the selected identity's version-stack shape follows disk order. * fix(ecstore): resolve selection ties by disk order and guard writes Review of the read-quorum selection found three gaps: - Rejecting a same-time half split made explicit-version reads of a partially applied tagging update fail and blocked the retry that repairs it. Resolve ties to the identity on the lowest-index disk, for reads and heal alike, instead of rejecting or using map order. - Delete markers and empty versions can lower the common read quorum below the selected version's data blocks, so HEAD could report a version GET cannot decode. Require that many copies, as MinIO's getLatestFileInfo does. - A selection can now return exactly read-quorum disks, and partial metadata writes are not rolled back. Refuse an update before writing when its target disks cannot reach write quorum. * fix(ecstore): count identity copies in the data-blocks guard The first-candidate shortcut counts byte-identical copies, so copies that differ only in identity-neutral fields such as replication status were missing from the data-blocks check. When the shortcut's count falls short, count every copy of the selected content identity before refusing the version. * fix(ecstore): keep commit-proof selection for system-bucket reads Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an ordinary read of pool.bin as proof that the record committed. Under the read-quorum rule, a COMMIT that reached only half of a 2+2 set was served as committed, and the pool_meta_native_physical_* integration tests failed. Route reads through select_fileinfo_for_read: S3 buckets keep MinIO's read-quorum rule, while is_meta_bucketname reads use the previous algorithm unchanged as select_committed_fileinfo, so a split stays unreadable until native recovery resolves it. --- .../src/set_disk/core/io_primitives.rs | 15 +- .../src/set_disk/core/metadata_quorum.rs | 6 +- crates/ecstore/src/set_disk/metadata.rs | 132 ++++-- crates/ecstore/src/set_disk/mod.rs | 380 +++++++++++++----- crates/ecstore/src/set_disk/ops/object.rs | 9 +- crates/ecstore/src/set_disk/read.rs | 28 +- crates/ecstore/src/store/list_objects.rs | 152 ++++++- docs/architecture/erasure-coding.md | 2 +- 8 files changed, 568 insertions(+), 156 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 99178c7de..a96e077ef 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -5752,6 +5752,19 @@ impl SetDisks { return Ok(()); } + // A metadata update that misses write quorum is not rolled back, so + // refuse it before writing when the target disks cannot reach quorum + // even if every write succeeds; otherwise the change a caller reports + // as failed would be served and spread by heal. + let write_quorum = fi.write_quorum(self.default_write_quorum()); + let reachable = disks + .iter() + .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound)) + .collect::>(); + if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) { + return Err(err); + } + self.invalidate_get_object_metadata_cache(bucket, object).await; let mut futures = Vec::with_capacity(disks.len()); @@ -5781,7 +5794,7 @@ impl SetDisks { } } - if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, fi.write_quorum(self.default_write_quorum())) { + if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) { return Err(err); } diff --git a/crates/ecstore/src/set_disk/core/metadata_quorum.rs b/crates/ecstore/src/set_disk/core/metadata_quorum.rs index d7c336b6d..d465cb722 100644 --- a/crates/ecstore/src/set_disk/core/metadata_quorum.rs +++ b/crates/ecstore/src/set_disk/core/metadata_quorum.rs @@ -597,10 +597,8 @@ mod tests { .iter() .map(|slot| slot.file_info().expect("every disk must have arrived").clone()) .collect::>(); - let (_, selected, selection_quorum) = - SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], "", 2, 3) - .expect("three matching B disks must select the current identity"); - assert_eq!(selection_quorum, 3); + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], 2) + .expect("three matching B disks must select the current identity"); assert_eq!(selected.version_id, expected_current.version_id); assert_eq!(selected.mod_time, expected_current.mod_time); assert_eq!(selected.data_dir, expected_current.data_dir); diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index d1c044c0b..01614136f 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -665,14 +665,20 @@ impl SetDisks { read_quorum } - pub(super) fn select_valid_fileinfo( + /// 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, 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); @@ -683,14 +689,71 @@ impl SetDisks { && 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)); + 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, 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 + /// itself reach `read_quorum`. Heal picks the latest version the same + /// way, so a read returns the version heal converges the set to; a write + /// that failed after reaching exactly `read_quorum` copies is readable + /// instead of failing until heal runs, as listing already shows it. + /// + /// `read_quorum` follows the parity most copies agree on, which delete + /// markers and empty versions can lower below the selected version's own + /// data blocks; like MinIO's `getLatestFileInfo`, a version with local + /// shards also needs that many copies, or HEAD would report a version + /// GET cannot decode. + pub(super) fn select_valid_fileinfo( + disks: &[Option], + parts_metadata: &[FileInfo], + errs: &[Option], + read_quorum: usize, + ) -> disk::error::Result<(Vec>, FileInfo)> { + let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, read_quorum); + let (fi, copies) = Self::find_file_info_in_quorum_counted(parts_metadata, &mod_time, &etag, read_quorum)?; + if !fi.deleted && !fi.is_remote() && fi.size > 0 && copies < fi.erasure.data_blocks { + // The shortcut counts byte-identical copies only; count every copy + // of the content identity before refusing the version. + let identity = Self::file_info_quorum_hash(&fi); + let copies = parts_metadata + .iter() + .filter(|meta| file_info_is_valid_for_metadata(meta) && Self::file_info_quorum_hash(meta) == identity) + .count(); + if copies < fi.erasure.data_blocks { + return Err(DiskError::ErasureReadQuorum); + } + } + + Ok((online_disks, fi)) } pub(super) fn pick_valid_fileinfo( @@ -1025,6 +1088,23 @@ impl SetDisks { etag: &Option, quorum: usize, ) -> disk::error::Result { + Self::find_file_info_in_quorum_counted(metas, mod_time, etag, quorum).map(|(fi, _)| fi) + } + + /// `find_file_info_in_quorum`, also returning how many copies carry the + /// selected content identity. + /// + /// Two identities can share the top count only in a data = parity set + /// split exactly in half at one modification time, such as a metadata + /// update that failed after reaching half the drives. The identity held + /// by the lowest-index disk wins, so every read and heal picks the same + /// side and a retried metadata write can repair the split. + fn find_file_info_in_quorum_counted( + metas: &[FileInfo], + mod_time: &Option, + etag: &Option, + quorum: usize, + ) -> disk::error::Result<(FileInfo, usize)> { if quorum < 1 { warn!("find_file_info_in_quorum: quorum < 1"); return Err(DiskError::ErasureReadQuorum); @@ -1053,7 +1133,7 @@ impl SetDisks { if same_count >= quorum && same_count > total_valid - same_count { let mut fi = first.clone(); fi.is_latest = fi.successor_mod_time.is_none(); - return Ok(fi); + return Ok((fi, same_count)); } } @@ -1106,15 +1186,12 @@ impl SetDisks { *count_map.entry(hash).or_insert(0) += 1; } - let mut max_val = None; - let mut max_count = 0; - - for (&val, &count) in &count_map { - if count > max_count { - max_val = Some(val); - max_count = count; - } - } + let max_count = count_map.values().copied().max().unwrap_or(0); + let max_val = meta_hashes + .iter() + .flatten() + .find(|hash| count_map.get(*hash) == Some(&max_count)) + .copied(); if max_count < quorum { warn!( @@ -1155,17 +1232,24 @@ impl SetDisks { if found { let mut fi = found_fi.expect("operation should succeed"); - for (val, &count) in &valid_obj_map { - if count >= quorum { - fi.successor_mod_time = val.successor_mod_time; - fi.num_versions = val.num_versions; - fi.is_latest = val.successor_mod_time.is_none(); - - break; - } + // Scan in disk order so that two version-stack shapes with equal + // counts resolve the same way on every read. + let quorum_props = metas + .iter() + .zip(meta_hashes.iter()) + .filter(|(meta, hash)| hash.is_some() && **hash == max_val && file_info_is_valid_for_metadata(meta)) + .map(|(meta, _)| ObjProps { + successor_mod_time: meta.successor_mod_time, + num_versions: meta.num_versions, + }) + .find(|props| valid_obj_map.get(props).is_some_and(|&count| count >= quorum)); + if let Some(props) = quorum_props { + fi.successor_mod_time = props.successor_mod_time; + fi.num_versions = props.num_versions; + fi.is_latest = props.successor_mod_time.is_none(); } - return Ok(fi); + return Ok((fi, max_count)); } warn!("find_file_info_in_quorum: fileinfo not found"); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8affa3650..39283654f 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -1790,7 +1790,7 @@ mod prepared_get_object_metadata_tests { assert!(original.is_complete()); let (original_metadata, original_errors, _) = original.into_legacy(); assert!(original_errors.iter().all(Option::is_none)); - let (_, generation_a, _) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, "", 2, 3) + let (_, generation_a) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, 2) .expect("generation A must have real metadata quorum"); assert!(!generation_a.inline_data(), "fixture must use non-inline shard files"); let order = bounded_metadata_fanout_order(bucket, &object, 4, 2); @@ -1851,8 +1851,8 @@ mod prepared_get_object_metadata_tests { assert!(replacement.is_complete()); let (replacement_metadata, replacement_errors, _) = replacement.into_legacy(); assert!(replacement_errors.iter().all(Option::is_none)); - let (online, generation_b, _) = - SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, "", 2, 3) + let (online, generation_b) = + SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, 2) .expect("replacement must have healthy real metadata quorum"); assert_eq!(online.iter().flatten().count(), 4); assert_ne!(generation_a.data_dir, generation_b.data_dir); @@ -6829,7 +6829,7 @@ impl SetDisks { let (metadata, errors) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false).await?; // Match delete_object_version_with_purge, independently of data parity or online inventory size. let quorum = self.set_drive_count / 2 + 1; - let (_, marker, _) = Self::select_valid_fileinfo(&disks, &metadata, &errors, version_id, quorum, quorum) + let (_, marker) = Self::select_valid_fileinfo(&disks, &metadata, &errors, quorum) .map_err(|error| to_object_err(error.into(), vec![bucket, object, version_id]))?; if !marker.deleted || marker.tier_free_version() || marker.version_id.unwrap_or_else(Uuid::nil) != requested_version { return Err(Error::DataMovementOverwriteErr( @@ -11969,41 +11969,121 @@ mod tests { } #[test] - fn test_latest_fileinfo_selection_quorum_requires_write_quorum_when_full_metadata_is_available() { + fn test_latest_fileinfo_selection_breaks_same_time_tie_in_disk_order() { 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 tagged = |index, tag: &str| { + let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index); + fi.metadata.insert("x-amz-tagging".to_string(), tag.to_string()); + fi + }; let errs = vec![None, None, None, None]; - let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3); + // A tagging write that failed after reaching half the drives. + for (metas, expected) in [ + (vec![tagged(1, "a=1"), tagged(2, "a=1"), tagged(3, "a=2"), tagged(4, "a=2")], "a=1"), + (vec![tagged(1, "a=2"), tagged(2, "a=1"), tagged(3, "a=1"), tagged(4, "a=2")], "a=2"), + ] { + // Each call hashes into fresh maps; a map-order choice would flip. + for _ in 0..32 { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("a same-time half split serves the side on the lowest-index disk"); + let healed = + SetDisks::find_file_info_in_quorum(&metas, &Some(mod_time), &None, 2).expect("heal resolves the same split"); - assert_eq!(quorum, 3); + assert_eq!(selected.metadata.get("x-amz-tagging").map(String::as_str), Some(expected)); + assert_eq!(healed.metadata.get("x-amz-tagging").map(String::as_str), Some(expected)); + } + } } #[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]; + fn test_latest_fileinfo_selection_requires_the_versions_own_data_blocks() { + let old_mod_time = OffsetDateTime::now_utc(); + let new_mod_time = old_mod_time + time::Duration::seconds(1); + let old = |index| quorum_test_fileinfo(old_mod_time, Uuid::from_u128(1), "part-etag-old", index); + let new = |index| { + let mut fi = quorum_test_fileinfo(new_mod_time, Uuid::from_u128(2), "part-etag-new", index); + fi.erasure.data_blocks = 3; + fi.erasure.parity_blocks = 1; + fi + }; + let 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); + // Read quorum two comes from the 2+2 copies; the newer 3+1 version + // wins the time tie but cannot be decoded from two shards. + let metas = vec![old(1), old(2), new(3), new(4)]; + let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2); + assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + + let metas = vec![old(1), new(2), new(3), new(4)]; + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("three copies decode a 3+1 version"); + assert_eq!(selected.mod_time, Some(new_mod_time)); + + // Replication-status drift does not change the content identity, so + // the drifted copy still counts toward the version's data blocks. + let replication_status_key = format!( + "{}{}", + rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::SUFFIX_REPLICATION_STATUS + ); + let mut drifted = new(4); + drifted + .metadata + .insert(replication_status_key, "target-a=PENDING;".to_string()); + let metas = vec![old(1), new(2), new(3), drifted]; + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("a replication-status drift still leaves three copies of the version"); + assert_eq!(selected.mod_time, Some(new_mod_time)); + } + + #[tokio::test] + async fn update_object_meta_below_write_quorum_writes_nothing() { + use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::storage_api_contracts::object::ObjectIO as _; + + let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(4, 2).await; + let (bucket, object) = ("metadata-write-quorum", "object"); + set.make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + set.put_object( + bucket, + object, + &mut crate::object_api::PutObjReader::from_vec(b"body".to_vec()), + &ObjectOptions::default(), + ) + .await + .expect("object should commit on every drive"); + let disks = set.get_disks_internal().await; + let (metadata, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false) + .await + .expect("committed metadata should be readable"); + let mut update = metadata[0].clone(); + update.metadata.insert("x-amz-meta-probe".to_string(), "written".to_string()); + + // A read that selected one side of a 2+2 split returns two disks. + let mut targets = disks.clone(); + targets[2] = None; + targets[3] = None; + let err = set + .update_object_meta(bucket, object, update, &targets) + .await + .expect_err("two of four disks cannot reach write quorum"); + + assert_eq!(err, DiskError::ErasureWriteQuorum); + let (after, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false) + .await + .expect("metadata should stay readable"); + assert!( + after.iter().all(|fi| !fi.metadata.contains_key("x-amz-meta-probe")), + "a refused update must not reach any disk" + ); } #[test] - fn test_latest_fileinfo_selection_quorum_keeps_read_quorum_for_partial_overwrite_with_read_error() { + fn test_latest_fileinfo_selection_keeps_older_majority_over_partial_overwrite() { 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(); @@ -12016,26 +12096,15 @@ mod tests { ]; 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) + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) .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() { + fn test_latest_fileinfo_selection_serves_newer_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(); @@ -12048,9 +12117,120 @@ mod tests { ]; let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; - let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + // The new version may hold write quorum through the missing disk; the + // old one cannot, and heal converges the set to the new version. + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("the newer version reaches read quorum"); - assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + assert_eq!(selected.data_dir, Some(new_data_dir)); + assert_eq!(selected.mod_time, Some(new_mod_time)); + } + + #[test] + fn test_latest_fileinfo_selection_serves_newer_side_of_complete_split() { + 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); + let errs = vec![None, None, None, None]; + + for metas in [vec![old(1), old(2), new(3), new(4)], vec![new(1), old(2), new(3), old(4)]] { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("a 2+2 split with every disk answering serves the newer version"); + + assert_eq!(selected.data_dir, Some(new_data_dir)); + assert_eq!(selected.parts[0].etag, "part-etag-new"); + assert!(selected.is_latest); + } + } + + #[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(); + let data_dir = Uuid::new_v4(); + let successor = mod_time + time::Duration::seconds(1); + let shape = |index, successor_mod_time: Option, num_versions| { + let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index); + fi.successor_mod_time = successor_mod_time; + fi.is_latest = successor_mod_time.is_none(); + fi.num_versions = num_versions; + fi + }; + let errs = vec![None, None, None, None]; + + for (metas, expected) in [ + ( + vec![ + shape(1, Some(successor), 2), + shape(2, None, 1), + shape(3, Some(successor), 2), + shape(4, None, 1), + ], + Some(successor), + ), + ( + vec![ + shape(1, None, 1), + shape(2, Some(successor), 2), + shape(3, None, 1), + shape(4, Some(successor), 2), + ], + None, + ), + ] { + // Each call hashes into fresh maps; a map-order choice would flip. + for _ in 0..32 { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("one content identity reaches read quorum"); + + assert_eq!(selected.successor_mod_time, expected); + assert_eq!(selected.is_latest, expected.is_none()); + assert_eq!(selected.num_versions, if expected.is_none() { 1 } else { 2 }); + } + } } #[test] @@ -12068,11 +12248,9 @@ mod tests { let metas = vec![first, second, FileInfo::default(), FileInfo::default()]; let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)]; - let (_, mut 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"); + let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .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"); assert!(selected.parts[0].checksums.is_none()); @@ -12099,7 +12277,7 @@ mod tests { let metas = vec![first, second, FileInfo::default(), FileInfo::default()]; let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)]; - let (_, mut selected, _) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3) + let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) .expect("winner selection should defer sidecar decoding"); let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected) .expect_err("a malformed degraded winner must fail closed when checksums are requested"); @@ -12166,10 +12344,9 @@ mod tests { 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) + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) .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)); @@ -12262,81 +12439,77 @@ mod tests { 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) + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) .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() { + fn test_latest_fileinfo_selection_serves_same_time_metadata_majority() { 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 colored = |index, color: &str| { + let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index); + fi.metadata.insert("x-amz-meta-color".to_string(), color.to_string()); + fi + }; 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); + // With one disk missing, only the two-copy side can hold write quorum, + // whichever disk order the copies arrive in. + for (metas, expected) in [ + ( + vec![colored(1, "blue"), colored(2, "blue"), colored(3, "red"), FileInfo::default()], + "blue", + ), + (vec![colored(1, "blue"), colored(2, "red"), colored(3, "red"), FileInfo::default()], "red"), + ] { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("the read-quorum side of a same-time metadata split is served"); - assert_eq!(quorum, 2); - assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + assert_eq!(selected.metadata.get("x-amz-meta-color").map(String::as_str), Some(expected)); + } } #[test] - fn test_latest_fileinfo_selection_rejects_same_modtime_partial_metadata_read_quorum() { + fn test_latest_fileinfo_selection_serves_same_time_transition_majority() { 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 local = |index| quorum_test_fileinfo(mod_time, data_dir, "part-etag", index); + let remote = |index| { + let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index); + fi.transition_status = TRANSITION_COMPLETE.to_string(); + fi.transition_tier = "WARM".to_string(); + fi.transitioned_objname = "remote/object".to_string(); + fi.transition_version_id = Some(Uuid::from_u128(7)); + fi + }; let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; - let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3); + for (metas, expect_remote) in [ + (vec![local(1), local(2), remote(3), FileInfo::default()], false), + (vec![local(1), remote(2), remote(3), FileInfo::default()], true), + ] { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("the read-quorum side of a same-time transition split is served"); - assert!(matches!(result, Err(DiskError::ErasureReadQuorum))); + assert_eq!(selected.is_remote(), expect_remote); + } + + for (tied, expect_remote) in [ + (vec![local(1), local(2), remote(3), remote(4)], false), + (vec![remote(1), local(2), local(3), remote(4)], true), + ] { + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; tied.len()], &tied, &[None, None, None, None], 2) + .expect("a same-time half split serves the side on the lowest-index disk"); + assert_eq!(selected.is_remote(), expect_remote); + } } #[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() { + fn test_latest_fileinfo_selection_serves_degraded_committed_identity() { let mod_time = OffsetDateTime::now_utc(); let data_dir = Uuid::new_v4(); let metas = vec![ @@ -12347,7 +12520,10 @@ mod tests { ]; let errs = vec![None, None, None, Some(DiskError::DiskNotFound)]; - assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3), 3); + let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2) + .expect("a committed identity stays readable with one disk missing"); + + assert_eq!(selected.parts[0].etag, "part-etag-a"); } #[test] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c5e088514..baede67c7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -9320,8 +9320,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { }; 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) + 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 { @@ -11182,8 +11182,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?; 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]))?; + 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 74cc091a4..e99512a10 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -966,8 +966,9 @@ impl SetDisks { return Err(to_object_err(err.into(), vec![bucket, object, &vid])); } - let (op_online_disks, mut fi, fileinfo_selection_quorum) = - match Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum) { + 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); @@ -987,7 +988,7 @@ impl SetDisks { rustfs_utils::http::remove_str(&mut metadata.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS); } } - metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, fileinfo_selection_quorum); + metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, read_quorum); let needs_late_metadata_fanout = !metadata_fanout_complete && allow_early_stop && non_inline_data_read_early_stop_allowed(read_data, bucket, object) @@ -1014,7 +1015,6 @@ impl SetDisks { && allow_early_stop && metadata_fanout_diagnostics.valid_responses() >= read_quorum && errs.iter().all(Option::is_none) - && fileinfo_selection_quorum >= read_quorum && op_online_disks.iter().filter(|disk| disk.is_some()).count() >= read_quorum)) { #[cfg(test)] @@ -2465,8 +2465,8 @@ impl SetDisks { 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, write_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])); } @@ -4052,8 +4052,7 @@ mod metadata_cache_tests { assert!(original.is_complete()); let (baseline, errors, _) = original.into_legacy(); assert!(errors.iter().all(Option::is_none), "original real shards must all be readable"); - let (_, selected, _) = - SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, "", 2, 3).expect("original metadata quorum"); + let (_, selected) = SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, 2).expect("original metadata quorum"); assert!(!selected.inline_data(), "late fixture must retain real non-inline shard files"); let expected = LateMetadataIdentity::from_file_info(&selected); let (_, _, original_online) = @@ -4086,7 +4085,7 @@ mod metadata_cache_tests { .expect("read changed real disk metadata"); assert!(changed.is_complete(), "late fixture must reach complete metadata observation"); let (metadata, errors, _) = changed.into_legacy(); - let (online, selected, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, "", 2, 3) + let (online, selected) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, 2) .expect("changed metadata still reaches selection quorum"); assert_eq!(online.iter().flatten().count(), 4, "the late identity guard must decide this failure"); if changed_generation { @@ -5135,7 +5134,7 @@ mod metadata_cache_tests { // This seam prepares a real, independently validated D1/P1/P2 quorum. // The hedge test separately exercises its production publication timing. let errs: Vec> = (0..4).map(|_| None).collect(); - let (online_disks, fi, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, "", 2, 3) + let (online_disks, fi) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, 2) .expect("three real members satisfy latest-generation selection"); assert_eq!(online_disks.iter().flatten().count(), 3); assert_eq!(fi.metadata.get("etag"), committed.etag.as_ref()); @@ -6240,16 +6239,17 @@ mod tests { let parts_metadata = vec![first.clone(), second, third, FileInfo::default()]; let errs = vec![None, None, None, None]; - let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2) + let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2) .expect("partial early-stop metadata should preserve read quorum"); let read_quorum = usize::try_from(read_quorum).expect("read quorum should be non-negative"); - let write_quorum = usize::try_from(write_quorum).expect("write quorum should be non-negative"); - let selection_quorum = SetDisks::latest_fileinfo_selection_quorum("", &parts_metadata, &errs, read_quorum, write_quorum); + let (_, latest) = SetDisks::select_valid_fileinfo(&vec![None; parts_metadata.len()], &parts_metadata, &errs, read_quorum) + .expect("partial early-stop metadata should select the candidate"); let selected = SetDisks::pick_valid_fileinfo(&parts_metadata, None, Some("etag-1".to_string()), read_quorum) .expect("partial early-stop metadata should preserve selected FileInfo"); assert_eq!(read_quorum, 2); - assert_eq!(selection_quorum, 3); + assert_eq!(latest.name, first.name); + assert_eq!(latest.get_etag(), first.get_etag()); assert_eq!(selected.name, first.name); assert_eq!(selected.get_etag(), first.get_etag()); } diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 13e4a3049..5469f4b31 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -6481,6 +6481,147 @@ mod test { } } + #[tokio::test] + async fn latest_get_follows_read_quorum_on_torn_writes() { + use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::storage_api_contracts::object::ObjectOperations as _; + + // HEAD and GET serve the version with the most copies once it reaches + // the object's data blocks, and a tie goes to the newer version, as in + // MinIO's getObjectFileInfo and as heal converges the set. GET used to + // demand write quorum whenever enough metadata answered, so a 2+2 or + // 8+8 split, or a newer side that could still hold write quorum + // through an offline drive, listed but answered 503. + let opts = ObjectOptions { + suppress_read_repair: true, + ..Default::default() + }; + for (drives, parity) in [(4_usize, 2_usize), (16, 8), (16, 4)] { + let data = drives - parity; + let bucket = format!("torn-write-get-{drives}-{parity}"); + let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(drives, parity).await; + set.make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + // Two committed generations of one key give every drive its own + // inline shard of each generation; the cases copy them per drive. + let mut generations = Vec::new(); + for body in [b"old generation".as_slice(), b"new generation".as_slice()] { + let info = set + .put_object( + &bucket, + "template", + &mut PutObjReader::from_vec(body.to_vec()), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("generation PUT should commit on every drive"); + let mut metadata = Vec::with_capacity(drives); + for dir in &dirs { + metadata.push( + tokio::fs::read(dir.path().join(&bucket).join("template").join(STORAGE_FORMAT_FILE)) + .await + .expect("generation metadata should carry the inline shard"), + ); + } + generations.push((body, info.etag.expect("PUT should report an etag"), metadata)); + } + let (new_body, new_etag, new) = generations.pop().expect("new generation"); + let (old_body, old_etag, old) = generations.pop().expect("old generation"); + + // New copies occupy the first drives, so the offline drive always + // removes one new copy. Each phase reads its own keys so the GET + // metadata cache never answers for the other phase. + let mut cases = Vec::new(); + for offline in [false, true] { + for copies in 1..drives { + for overwrite in [false, true] { + let name = format!( + "{}-{}-{copies:02}", + if offline { "offline" } else { "online" }, + if overwrite { "overwrite" } else { "new" } + ); + for slot in 0..drives { + let bytes = if slot < copies { + &new[slot] + } else if overwrite { + &old[slot] + } else { + continue; + }; + let path = dirs[slot].path().join(&bucket).join(&name); + tokio::fs::create_dir_all(&path).await.expect("create object directory"); + tokio::fs::write(path.join(STORAGE_FORMAT_FILE), bytes) + .await + .expect("write generation metadata"); + } + cases.push((offline, name, copies, overwrite)); + } + } + } + + for offline in [false, true] { + let parked = if offline { set.disks.write().await[0].take() } else { None }; + for (_, name, copies, overwrite) in cases.iter().filter(|case| case.0 == offline) { + let new_copies = copies - usize::from(offline); + let old_copies = if *overwrite { drives - copies } else { 0 }; + let expected = if new_copies >= data && new_copies >= old_copies { + Some((&new_etag, new_body)) + } else if old_copies >= data { + Some((&old_etag, old_body)) + } else { + None + }; + let context = format!("{drives} drives EC {data}+{parity}, {name}: {new_copies} new, {old_copies} old"); + let head = set.get_object_info(&bucket, name, &opts).await; + let get = set.get_object_reader(&bucket, name, None, HeaderMap::new(), &opts).await; + match (expected, head, get) { + (Some((etag, body)), Ok(info), Ok(mut reader)) => { + assert_eq!(info.etag.as_ref(), Some(etag), "{context}: HEAD"); + assert_eq!(reader.object_info.etag.as_ref(), Some(etag), "{context}: GET"); + let mut streamed = Vec::new(); + reader + .stream + .read_to_end(&mut streamed) + .await + .unwrap_or_else(|err| panic!("{context}: GET body: {err:?}")); + assert_eq!(streamed, body, "{context}: GET body"); + } + (Some((etag, _)), head, get) => panic!( + "{context}: expected {etag}, HEAD {:?}, GET {:?}", + head.map(|info| info.etag), + get.map(|reader| reader.object_info.etag) + ), + (None, head, get) => { + for (method, result) in [ + ("HEAD", head.map(|info| info.etag)), + ("GET", get.map(|reader| reader.object_info.etag)), + ] { + let err = result.expect_err(&format!("{context}: {method} expected 404 or unavailable")); + assert!( + matches!( + err, + StorageError::ObjectNotFound(..) + | StorageError::FileNotFound + | StorageError::ErasureReadQuorum + | StorageError::InsufficientReadQuorum(..) + ), + "{context}: {method} expected 404 or unavailable, got {err:?}" + ); + } + } + } + } + if offline { + set.disks.write().await[0] = parked; + } + } + } + } + #[tokio::test] async fn latest_listing_and_get_agree_on_hermetic_set_states() { #[derive(Clone, Copy)] @@ -6553,6 +6694,10 @@ mod test { ("d-torn-two", [Some(Generation::New), Some(Generation::New), None, None]), ("e-torn-one", [Some(Generation::New), None, None, None]), ("f-stale-one", [None, None, None, Some(Generation::Old)]), + ( + "fa-overwrite-two-of-three", + [Some(Generation::New), Some(Generation::New), Some(Generation::Old), None], + ), ( "g-overwrite-torn-one", [ @@ -6597,6 +6742,7 @@ mod test { ("b-copied", &new_etag), ("c-split", &new_etag), ("d-torn-two", &new_etag), + ("fa-overwrite-two-of-three", &new_etag), ("g-overwrite-torn-one", &old_etag), ] .into_iter() @@ -6623,12 +6769,6 @@ mod test { "{name}: GET must serve the version LIST selected" ); } - // Known divergence: when every drive answers, GET of the latest - // version demands write quorum (three of four), so a 2+2 split - // answers 503 although LIST shows the newer generation, as - // MinIO's listing does. F.0's dangling classification does not - // cover a split; a GET change is a separate decision. - (Some(_), Err(StorageError::InsufficientReadQuorum(..))) if name == "c-split" => {} (Some(_), Err(err)) => assert!( matches!( err, diff --git a/docs/architecture/erasure-coding.md b/docs/architecture/erasure-coding.md index b4cb3351f..6b8c04a84 100644 --- a/docs/architecture/erasure-coding.md +++ b/docs/architecture/erasure-coding.md @@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli ## 8. Read path and read quorum - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards. -- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Latest-version reads may escalate to write_quorum to avoid resurrecting a partially-overwritten version. +- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13). - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued. - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).