From 9d6282e95d8a367ad18a665f43d87806ad40f33b Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 15:28:19 +0800 Subject: [PATCH] fix(ecstore): guard latest rebalance tie-breaks --- crates/ecstore/src/store/rebalance.rs | 209 +++++++++++++++++- crates/ecstore/src/store/rebalance/support.rs | 87 ++++++-- 2 files changed, 275 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 7d7c1d91e..4ad030a0c 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -1423,6 +1423,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 +1454,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 +1474,205 @@ 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_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 6035e58d8..2efc0bffd 100644 --- a/crates/ecstore/src/store/rebalance/support.rs +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -137,37 +137,84 @@ 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 +} + +/// Pool-specific erasure geometry is intentionally excluded. All fields that +/// identify the selected object version and its payload 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.size == right.size + && left.actual_size == right.actual_size + && left.is_dir == right.is_dir + && left.version_id == right.version_id + && left.data_dir == right.data_dir + && left.delete_marker == right.delete_marker + && left.restore_ongoing == right.restore_ongoing + && left.restore_expires == right.restore_expires + && left.user_tags == right.user_tags + && left.content_type == right.content_type + && left.content_encoding == right.content_encoding + && left.expires == right.expires + && left.etag == right.etag + && left.inlined == right.inlined + && left.parts == right.parts + && left.checksum == right.checksum + && same_transition_identity(left, right) +} + 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)