mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-23 12:49:04 +00:00
fix(ecstore): guard latest rebalance tie-breaks
This commit is contained in:
overtrue
@@ -1423,6 +1423,14 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn object_info_with_identity(unix_ts: i64, delete_marker: bool, version_id: Uuid, etag: Option<String>) -> 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(
|
||||
|
||||
@@ -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<OffsetDateTime> {
|
||||
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<LatestObjectInfoCandidate>,
|
||||
candidates: Vec<LatestObjectInfoCandidate>,
|
||||
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::<Vec<_>>();
|
||||
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user