fix(ecstore): tolerate migration identity rewrites (#6447)

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,

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() {

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()));

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(