From 26e6508b64401131127a35aebede0d56ffa34507 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 23 Aug 2026 01:42:11 +0800 Subject: [PATCH] fix(ecstore): reject equal-time latest identity conflicts before index fallback (#6374) --- crates/ecstore/src/store/rebalance.rs | 387 +++++++++++++++++- crates/ecstore/src/store/rebalance/support.rs | 167 +++++++- 2 files changed, 532 insertions(+), 22 deletions(-) diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 7d7c1d91e..dc2fa2ee5 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -859,6 +859,7 @@ fn lifecycle_delete_all_test_failure(phase: crate::object_api::LifecycleDeleteAl #[cfg(test)] mod tests { use super::*; + use crate::bucket::replication::{ReplicationStatusType, VersionPurgeStatusType}; use crate::config::storageclass::{CLASS_RRS, CLASS_STANDARD, lookup_config_for_pools_without_env}; use crate::disk::error::DiskError; use crate::layout::endpoint::Endpoint; @@ -1423,6 +1424,14 @@ mod tests { } } + fn object_info_with_identity(unix_ts: i64, delete_marker: bool, version_id: Uuid, etag: Option) -> ObjectInfo { + ObjectInfo { + version_id: Some(version_id), + etag, + ..object_info_with_mod_time(unix_ts, delete_marker) + } + } + #[test] fn resolve_latest_object_info_candidates_returns_latest_delete_marker() { let candidates = vec![ @@ -1446,7 +1455,7 @@ mod tests { } #[test] - fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time() { + fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time_for_equivalent_candidates() { let candidates = vec![ LatestObjectInfoCandidate { info: Some(object_info_with_mod_time(10, false)), @@ -1466,6 +1475,382 @@ mod tests { assert_eq!(idx, 1); } + #[test] + fn resolve_latest_object_info_candidates_keeps_index_fallback_for_fully_equivalent_identities() { + let candidates = vec![ + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))), + idx: 2, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))), + idx: 7, + err: None, + }, + ]; + + let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()) + .expect("equivalent replicas must resolve deterministically"); + + assert_eq!(idx, 7); + assert_eq!(info.version_id, Some(Uuid::from_u128(1))); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_equal_time_version_id_conflict() { + let candidates = vec![ + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(2), Some("etag-a".to_string()))), + idx: 1, + err: None, + }, + ]; + + let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()) + .expect_err("divergent version ids must not silently resolve to the higher pool index"); + + assert_eq!(err, Error::ErasureReadQuorum); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_equal_time_etag_conflict() { + let candidates = vec![ + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-old".to_string()))), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-new".to_string()))), + idx: 1, + err: None, + }, + ]; + + let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()) + .expect_err("divergent etags must not silently resolve to the higher pool index"); + + assert_eq!(err, Error::ErasureReadQuorum); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_equal_time_delete_marker_conflict() { + let candidates = vec![ + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), None)), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, true, Uuid::from_u128(1), Some("etag-a".to_string()))), + idx: 1, + err: None, + }, + ]; + + let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()) + .expect_err("a delete marker tied with a live version must not be masked by the pool index"); + + assert_eq!(err, Error::ErasureReadQuorum); + } + + fn assert_equal_time_identity_conflict(left: ObjectInfo, right: ObjectInfo) { + let err = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(left), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(right), + idx: 1, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect_err("equal-time identity divergence must fail closed"); + + assert_eq!(err, Error::ErasureReadQuorum); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_equal_time_payload_identity_conflicts() { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + + let mut data_dir = base.clone(); + data_dir.data_dir = Some(Uuid::from_u128(2)); + assert_equal_time_identity_conflict(base.clone(), data_dir); + + let mut size = base.clone(); + size.size = 1; + assert_equal_time_identity_conflict(base.clone(), size); + + let mut actual_size = base.clone(); + actual_size.actual_size = 1; + assert_equal_time_identity_conflict(base.clone(), actual_size); + + let mut checksum = base.clone(); + checksum.checksum = Some(bytes::Bytes::from_static(b"checksum")); + assert_equal_time_identity_conflict(base.clone(), checksum); + + let mut parts = base.clone(); + parts.parts = std::sync::Arc::new(vec![rustfs_filemeta::ObjectPartInfo { + etag: "part-etag".to_string(), + number: 1, + size: 1, + ..Default::default() + }]); + assert_equal_time_identity_conflict(base.clone(), parts); + + let mut transition = base; + transition.transitioned_object.tier = "tier-a".to_string(); + assert_equal_time_identity_conflict( + object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())), + transition, + ); + } + + #[test] + fn resolve_latest_object_info_candidates_accepts_internal_metadata_aliases() { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + let mut rustfs_alias = base.clone(); + rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + "x-rustfs-internal-compression".to_string(), + "zstd".to_string(), + )])); + let mut minio_alias = base.clone(); + minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + "X-MINIO-INTERNAL-COMPRESSION".to_string(), + "zstd".to_string(), + )])); + + let (_, idx) = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(rustfs_alias), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(minio_alias), + idx: 1, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect("same-value internal aliases should resolve"); + assert_eq!(idx, 1); + + let mut dual_alias = base.clone(); + dual_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([ + ("x-rustfs-internal-compression".to_string(), "zstd".to_string()), + ("x-minio-internal-compression".to_string(), "zstd".to_string()), + ])); + let mut single_alias = base; + single_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + "x-rustfs-internal-compression".to_string(), + "zstd".to_string(), + )])); + + let (_, idx) = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(dual_alias), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(single_alias), + idx: 1, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect("dual-key and single-key internal metadata should resolve"); + assert_eq!(idx, 1); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_different_internal_metadata_alias_values() { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + let mut rustfs_alias = base.clone(); + rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + "x-rustfs-internal-compression".to_string(), + "zstd".to_string(), + )])); + let mut minio_alias = base; + minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + "x-minio-internal-compression".to_string(), + "snappy".to_string(), + )])); + + assert_equal_time_identity_conflict(rustfs_alias, minio_alias); + } + + #[test] + fn resolve_latest_object_info_candidates_preserves_dynamic_internal_metadata_identity_case() { + for suffix_prefix in ["replication-reset-", "replication-delete-marker-version-"] { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + let mut rustfs_alias = base.clone(); + rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + format!( + "X-RUSTFS-INTERNAL-{}{suffix}", + suffix_prefix.to_uppercase(), + suffix = "arn:aws:s3:::Bucket" + ), + "value".to_string(), + )])); + let mut minio_alias = base.clone(); + minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + format!("x-minio-internal-{suffix_prefix}arn:aws:s3:::Bucket"), + "value".to_string(), + )])); + + let (_, idx) = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(rustfs_alias.clone()), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(minio_alias), + idx: 1, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect("dynamic internal aliases with the same target should resolve"); + assert_eq!(idx, 1); + + let mut different_target_case = base; + different_target_case.user_defined = std::sync::Arc::new(std::collections::HashMap::from([( + format!("x-minio-internal-{suffix_prefix}arn:aws:s3:::bucket"), + "value".to_string(), + )])); + + assert_equal_time_identity_conflict(rustfs_alias, different_target_case); + } + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_conflicting_internal_metadata_aliases_in_one_candidate() { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + let mut first = base.clone(); + first.user_defined = std::sync::Arc::new(std::collections::HashMap::from([ + ("x-rustfs-internal-compression".to_string(), "zstd".to_string()), + ("x-minio-internal-compression".to_string(), "snappy".to_string()), + ])); + let mut second = base; + second.user_defined = first.user_defined.clone(); + + assert_equal_time_identity_conflict(first, second); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_replication_identity_conflict() { + let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + + let mut replication = base.clone(); + replication.replication_status_internal = Some("PENDING".to_string()); + replication.replication_status = ReplicationStatusType::Pending; + assert_equal_time_identity_conflict(base.clone(), replication); + + let mut purge = base.clone(); + purge.version_purge_status_internal = Some("PENDING".to_string()); + purge.version_purge_status = VersionPurgeStatusType::Pending; + assert_equal_time_identity_conflict(base.clone(), purge); + + let mut decision = base; + decision.replication_decision = "replicate".to_string(); + assert_equal_time_identity_conflict( + object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())), + decision, + ); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_none_vs_unix_epoch_mod_time() { + let mut without_mod_time = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string())); + without_mod_time.mod_time = None; + let with_unix_epoch = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string())); + + assert_equal_time_identity_conflict(without_mod_time, with_unix_epoch); + } + + #[test] + fn resolve_latest_object_info_candidates_ignores_older_identity_conflicts() { + let latest = object_info_with_identity(20, false, Uuid::from_u128(1), Some("etag-latest".to_string())); + let mut older = object_info_with_identity(10, true, Uuid::from_u128(2), Some("etag-old".to_string())); + older.data_dir = Some(Uuid::from_u128(2)); + + let (info, idx) = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(latest), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(older), + idx: 9, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect("older identity divergence must not affect the latest candidate"); + + assert_eq!(idx, 0); + assert_eq!( + info.mod_time, + Some(OffsetDateTime::from_unix_timestamp(20).expect("operation should succeed")) + ); + } + + #[test] + fn resolve_latest_object_info_candidates_ignores_not_found_pools_when_resolving() { + let candidates = vec![ + LatestObjectInfoCandidate { + info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: None, + idx: 1, + err: Some(Error::ObjectNotFound("bucket".to_string(), "object".to_string())), + }, + ]; + + let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()) + .expect("not-found pools must not block resolution of found candidates"); + + assert_eq!(idx, 0); + assert_eq!(info.version_id, Some(Uuid::from_u128(1))); + } + #[test] fn resolve_latest_object_info_candidates_returns_non_not_found_error() { let err = resolve_latest_object_info_candidates( diff --git a/crates/ecstore/src/store/rebalance/support.rs b/crates/ecstore/src/store/rebalance/support.rs index e37fc97bc..6e4db41b8 100644 --- a/crates/ecstore/src/store/rebalance/support.rs +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -12,10 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::cmp::Ordering; +use std::collections::HashMap; use crate::error::{Error, Result, StorageError, is_err_object_not_found, is_err_version_not_found}; use crate::object_api::{ObjectInfo, ObjectOptions}; +use rustfs_utils::http::metadata_compat::{ + SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX, SUFFIX_REPLICATION_RESET_ARN_PREFIX, + strip_internal_prefix_preserving_case, +}; use rustfs_utils::path::decode_dir_object; use time::OffsetDateTime; @@ -137,37 +141,158 @@ pub(super) fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, p )) } +fn latest_candidate_mod_time(candidate: &LatestObjectInfoCandidate) -> Option { + candidate + .info + .as_ref() + .map(|info| info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)) +} + +fn same_transition_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { + left.transition_version_state == right.transition_version_state + && left.transitioned_object.name == right.transitioned_object.name + && left.transitioned_object.version_id == right.transitioned_object.version_id + && left.transitioned_object.tier == right.transitioned_object.tier + && left.transitioned_object.free_version == right.transitioned_object.free_version + && left.transitioned_object.status == right.transitioned_object.status +} + +#[derive(PartialEq, Eq)] +struct LatestUserDefinedIdentity { + internal: HashMap, + other: HashMap, +} + +fn normalize_internal_identity_suffix(key: &str) -> Option { + let suffix = strip_internal_prefix_preserving_case(key)?; + + for dynamic_prefix in [ + SUFFIX_REPLICATION_RESET_ARN_PREFIX, + SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX, + ] { + let prefix_len = dynamic_prefix.len(); + if let (Some(prefix), Some(remainder)) = (suffix.get(..prefix_len), suffix.get(prefix_len..)) + && prefix.eq_ignore_ascii_case(dynamic_prefix) + { + return Some(format!("{dynamic_prefix}{remainder}")); + } + } + + Some(suffix.to_lowercase()) +} + +fn normalize_user_defined_identity(user_defined: &HashMap) -> Option { + let mut identity = LatestUserDefinedIdentity { + internal: HashMap::with_capacity(user_defined.len()), + other: HashMap::with_capacity(user_defined.len()), + }; + + for (key, value) in user_defined { + if let Some(suffix) = normalize_internal_identity_suffix(key) { + if identity + .internal + .insert(suffix, value.clone()) + .is_some_and(|previous| previous != *value) + { + return None; + } + } else { + identity.other.insert(key.clone(), value.clone()); + } + } + + Some(identity) +} + +fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { + match ( + normalize_user_defined_identity(&left.user_defined), + normalize_user_defined_identity(&right.user_defined), + ) { + (Some(left), Some(right)) => left == right, + _ => false, + } +} + +/// Pool-specific erasure geometry is intentionally excluded: `get_object_info` +/// returns each pool's own `data_blocks`/`parity_blocks`, so those values can +/// differ for the same object version while the selected winner still carries +/// the chosen pool's layout. `put_object_reader` is also intentionally +/// excluded because it is a transient request handle that `ObjectInfo::clone` +/// drops. Every other ObjectInfo field is part of the production-visible +/// identity and must agree before the pool index can provide a deterministic +/// tie-break. +fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { + left.bucket == right.bucket + && left.name == right.name + && left.storage_class == right.storage_class + && left.mod_time == right.mod_time + && left.size == right.size + && left.actual_size == right.actual_size + && left.is_dir == right.is_dir + && same_user_defined_identity(left, right) + && left.user_tags == right.user_tags + && left.version_id == right.version_id + && left.data_dir == right.data_dir + && left.delete_marker == right.delete_marker + && same_transition_identity(left, right) + && left.restore_ongoing == right.restore_ongoing + && left.restore_expires == right.restore_expires + && left.parts == right.parts + && left.is_latest == right.is_latest + && left.content_type == right.content_type + && left.content_encoding == right.content_encoding + && left.expires == right.expires + && left.num_versions == right.num_versions + && left.successor_mod_time == right.successor_mod_time + && left.etag == right.etag + && left.inlined == right.inlined + && left.metadata_only == right.metadata_only + && left.version_only == right.version_only + && left.replication_status_internal == right.replication_status_internal + && left.replication_status == right.replication_status + && left.version_purge_status_internal == right.version_purge_status_internal + && left.version_purge_status == right.version_purge_status + && left.replication_decision == right.replication_decision + && left.checksum == right.checksum +} + pub(super) fn resolve_latest_object_info_candidates( - mut candidates: Vec, + candidates: Vec, bucket: &str, object: &str, opts: &ObjectOptions, ) -> Result<(ObjectInfo, usize)> { - candidates.sort_by(|a, b| { - let a_mod = if let Some(info) = &a.info { - info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH + let latest_mod_time = candidates.iter().filter_map(latest_candidate_mod_time).max(); + + if let Some(latest_mod_time) = latest_mod_time { + let mut latest_candidates = candidates + .into_iter() + .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time)) + .collect::>(); + + latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx)); + + let Some(winner) = latest_candidates.first() else { + return Err(Error::ErasureReadQuorum); + }; + let Some(winner_info) = winner.info.as_ref() else { + return Err(Error::ErasureReadQuorum); }; - let b_mod = if let Some(info) = &b.info { - info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH - }; - - if a_mod == b_mod { - return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less }; + if latest_candidates.iter().skip(1).any(|candidate| { + candidate + .info + .as_ref() + .is_none_or(|info| !same_latest_object_info_identity(winner_info, info)) + }) { + return Err(Error::ErasureReadQuorum); } - b_mod.cmp(&a_mod) - }); + return Ok((winner_info.clone(), winner.idx)); + } for candidate in candidates { - if let Some(info) = candidate.info { - return Ok((info, candidate.idx)); - } - if let Some(err) = candidate.err && !is_err_object_not_found(&err) && !is_err_version_not_found(&err)