From 43a10e3c244a85f338de2bda67d4f67ccf08ef89 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 23 Aug 2026 20:24:04 +0800 Subject: [PATCH] fix(ecstore): tolerate migration identity rewrites (#6447) --- crates/ecstore/src/data_movement/mod.rs | 2 +- crates/ecstore/src/store/init.rs | 89 +++++++++++++++++++ crates/ecstore/src/store/rebalance.rs | 56 +++++++++++- crates/ecstore/src/store/rebalance/support.rs | 66 +++++++++----- 4 files changed, 185 insertions(+), 28 deletions(-) diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 4fdda942e..03680ea45 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -813,7 +813,7 @@ pub(crate) fn is_equivalent_data_movement_metadata( .all(|(key, value)| source.user_defined.get(key) == Some(value)) } -fn is_equivalent_data_movement_object_identity( +pub(crate) fn is_equivalent_data_movement_object_identity( source: &ObjectInfo, target: &ObjectInfo, compare_mod_time: bool, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 5b9e67ab0..eb8269733 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -4266,6 +4266,95 @@ mod tests { assert_eq!(selected_body, conflicting_body); } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn data_movement_target_remains_readable_before_source_cleanup() { + let temp_dir = tempfile::tempdir().expect("create data movement read-window store dir"); + let (_ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "data-movement-read-window", &[4, 4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("dm-read-window-{}", uuid::Uuid::new_v4()); + let object = "object.bin"; + let version = uuid::Uuid::new_v4(); + let mod_time = OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create data movement read-window bucket"); + + let mut reader = PutObjReader::from_vec(b"source body".to_vec()); + store.pools[0] + .put_object( + &bucket, + object, + &mut reader, + &ObjectOptions { + versioned: true, + version_id: Some(version.to_string()), + mod_time: Some(mod_time), + ..Default::default() + }, + ) + .await + .expect("seed data movement source"); + *store.rebalance_meta.write().await = Some(active_rebalance_meta_for_pool(store.pools.len(), 0)); + + let source_reader = store.pools[0] + .get_object_reader( + &bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + versioned: true, + version_id: Some(version.to_string()), + raw_data_movement_read: true, + ..Default::default() + }, + ) + .await + .expect("read data movement source"); + let source_data_dir = source_reader.object_info.data_dir; + crate::data_movement::migrate_object(store.clone(), 0, bucket.clone(), source_reader, None, "test_read_window") + .await + .expect("commit data movement target"); + + let target = store.pools[1] + .get_object_info( + &bucket, + object, + &ObjectOptions { + versioned: true, + version_id: Some(version.to_string()), + ..Default::default() + }, + ) + .await + .expect("read committed data movement target"); + assert_ne!(source_data_dir, target.data_dir, "the target must use its own data directory"); + assert_eq!( + rustfs_utils::http::get_consistent_str(&target.user_defined, rustfs_utils::http::SUFFIX_DATA_MOVED), + Some("true") + ); + + let resolved = store + .get_object_info( + &bucket, + object, + &ObjectOptions { + versioned: true, + version_id: Some(version.to_string()), + ..Default::default() + }, + ) + .await + .expect("the equal-time migration window must remain readable"); + assert_eq!(resolved.data_dir, target.data_dir); + + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn data_movement_multipart_conflict_validates_exact_version_target_pool() { diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 550bd36e6..fb387db99 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -1638,10 +1638,6 @@ mod tests { 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); @@ -1671,6 +1667,58 @@ mod tests { ); } + #[test] + fn resolve_latest_object_info_candidates_accepts_data_movement_rewrites() { + let mut source = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + source.data_dir = Some(Uuid::from_u128(1)); + + let mut target = source.clone(); + target.data_dir = Some(Uuid::from_u128(2)); + rustfs_utils::http::insert_str( + Arc::make_mut(&mut target.user_defined), + rustfs_utils::http::SUFFIX_DATA_MOVED, + "true".to_string(), + ); + rustfs_utils::http::insert_str( + Arc::make_mut(&mut target.user_defined), + rustfs_utils::http::SUFFIX_ACTUAL_SIZE, + "0".to_string(), + ); + + let (info, idx) = resolve_latest_object_info_candidates( + vec![ + LatestObjectInfoCandidate { + info: Some(source), + idx: 0, + err: None, + }, + LatestObjectInfoCandidate { + info: Some(target.clone()), + idx: 1, + err: None, + }, + ], + "bucket", + "object", + &ObjectOptions::default(), + ) + .expect("a committed data-movement target must remain readable before source cleanup"); + + assert_eq!(idx, 1); + assert_eq!(info.data_dir, target.data_dir); + } + + #[test] + fn resolve_latest_object_info_candidates_rejects_unmarked_data_dir_conflict() { + let mut left = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())); + left.data_dir = Some(Uuid::from_u128(1)); + + let mut right = left.clone(); + right.data_dir = Some(Uuid::from_u128(2)); + + assert_equal_time_identity_conflict(left, right); + } + #[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())); diff --git a/crates/ecstore/src/store/rebalance/support.rs b/crates/ecstore/src/store/rebalance/support.rs index 434f9c29d..9717aec66 100644 --- a/crates/ecstore/src/store/rebalance/support.rs +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -214,47 +214,67 @@ fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool { } } -/// 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. +/// Pool-specific erasure geometry and data-movement rewrites are intentionally +/// excluded. The selected winner still carries the chosen pool's layout, while +/// the remaining read-visible fields 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 + let same_read_surface = left.bucket == right.bucket && left.name == right.name - && left.storage_class == right.storage_class + && left.is_dir == right.is_dir + && left.restore_ongoing == right.restore_ongoing + && left.restore_expires == right.restore_expires + && left.is_latest == right.is_latest + && left.content_type == right.content_type + && left.content_encoding == right.content_encoding + && left.num_versions == right.num_versions + && left.successor_mod_time == right.successor_mod_time + && left.inlined == right.inlined + && left.metadata_only == right.metadata_only + && left.version_only == right.version_only + && left.replication_decision == right.replication_decision; + + if !same_read_surface { + return false; + } + + let exact_identity = 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 + && left.checksum == right.checksum; + + if exact_identity { + return true; + } + + let left_moved = + rustfs_utils::http::get_consistent_str(&left.user_defined, rustfs_utils::http::SUFFIX_DATA_MOVED) == Some("true"); + let right_moved = + rustfs_utils::http::get_consistent_str(&right.user_defined, rustfs_utils::http::SUFFIX_DATA_MOVED) == Some("true"); + + match (left_moved, right_moved) { + (false, false) => false, + (false, true) => crate::data_movement::is_equivalent_data_movement_object_identity(left, right, true, true), + (true, false) => crate::data_movement::is_equivalent_data_movement_object_identity(right, left, true, true), + (true, true) => { + crate::data_movement::is_equivalent_data_movement_object_identity(left, right, true, true) + || crate::data_movement::is_equivalent_data_movement_object_identity(right, left, true, true) + } + } } pub(super) fn resolve_latest_object_info_candidates(