diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 118d66b9c..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); } @@ -9440,7 +9453,7 @@ mod tests { "lifecycle-inline-data-get-fanout-bucket", "lifecycle-inline-data-get-object", GET_OBJECT_PATH_LEGACY_DUPLEX, - None, + false, ); } @@ -9451,7 +9464,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 +9475,7 @@ mod tests { RUSTFS_META_BUCKET, "buckets/.usage-cache/lifecycle-inline-data-get-object", GET_OBJECT_PATH_INTERNAL_META, - None, + false, ); } @@ -9563,7 +9576,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 +9596,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 +9637,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") 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)).