fix(rebalance): converge multipart data movement retries (#6057)

* fix(rebalance): converge multipart data movement retries

* fix(rebalance): harden multipart retry replacement

* fix(rebalance): isolate internal multipart uploads

* test(ecstore): adapt metadata mutation fixtures

* fix(rebalance): preserve transition metadata semantics

* refactor(ecstore): reuse internal metadata matcher

* Revert "refactor(ecstore): reuse internal metadata matcher"

This reverts commit c87ca0328f.

* refactor(rebalance): reuse data movement log constants

* fix(rebalance): isolate migration-owned state

* fix(rebalance): preserve pre-gate retry compatibility

crates/ecstore/src/store/bucket_fence.rs

@@ -161,6 +161,10 @@ impl BucketIncarnationFenceGuard {
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
     }
+
+    pub(crate) fn namespace_lock_guard(&self) -> Option<&NamespaceLockGuard> {
+        self.inner.as_ref()
+    }
 }
 
 impl Drop for BucketIncarnationFenceGuard {

crates/ecstore/src/store/multipart.rs

@@ -66,6 +66,76 @@ fn ensure_multipart_bucket_lifecycle_guard_held(
     Ok(())
 }
 
+#[cfg(test)]
+struct DataMovementMultipartCompletionBarrierState {
+    bucket: String,
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+pub(crate) struct DataMovementMultipartCompletionBarrier {
+    state: Arc<DataMovementMultipartCompletionBarrierState>,
+}
+
+#[cfg(test)]
+static DATA_MOVEMENT_MULTIPART_COMPLETION_BARRIER: std::sync::OnceLock<
+    std::sync::Mutex<Option<Arc<DataMovementMultipartCompletionBarrierState>>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl DataMovementMultipartCompletionBarrier {
+    pub(crate) fn install(bucket: &str) -> Self {
+        let state = Arc::new(DataMovementMultipartCompletionBarrierState {
+            bucket: bucket.to_string(),
+            arrived: tokio::sync::Notify::new(),
+            release: tokio::sync::Notify::new(),
+        });
+        let mut slot = DATA_MOVEMENT_MULTIPART_COMPLETION_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("data movement multipart completion barrier mutex should not poison");
+        assert!(slot.is_none(), "data movement multipart completion barrier must be unique");
+        *slot = Some(Arc::clone(&state));
+        Self { state }
+    }
+
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(std::time::Duration::from_secs(30), self.state.arrived.notified())
+            .await
+            .expect("data movement multipart operation should reach selected completion");
+    }
+}
+
+#[cfg(test)]
+impl Drop for DataMovementMultipartCompletionBarrier {
+    fn drop(&mut self) {
+        self.state.release.notify_one();
+        let mut slot = DATA_MOVEMENT_MULTIPART_COMPLETION_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("data movement multipart completion barrier mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+async fn pause_data_movement_multipart_before_selected_completion(bucket: &str) {
+    let barrier = DATA_MOVEMENT_MULTIPART_COMPLETION_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("data movement multipart completion barrier mutex should not poison")
+        .as_ref()
+        .filter(|barrier| barrier.bucket == bucket)
+        .cloned();
+    if let Some(barrier) = barrier {
+        barrier.arrived.notify_one();
+        barrier.release.notified().await;
+    }
+}
+
 async fn list_pool_multipart_uploads_for_incarnation(
     pool: &crate::core::sets::Sets,
     bucket: &str,
@@ -332,7 +402,7 @@ impl ECStore {
     ) -> Result<MultipartUploadResult> {
         self.handle_new_multipart_upload_with_pool_idx(bucket, object, opts)
             .await
-            .map(|(res, _)| res)
+            .map(|(res, _, _)| res)
     }
 
     pub(crate) async fn handle_new_multipart_upload_with_pool_idx(
@@ -340,7 +410,7 @@ impl ECStore {
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
-    ) -> Result<(MultipartUploadResult, usize)> {
+    ) -> Result<(MultipartUploadResult, usize, Option<Uuid>)> {
         check_new_multipart_args(bucket, object)?;
         let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
         let opts = &opts;
@@ -349,7 +419,20 @@ impl ECStore {
             return self.pools[0]
                 .new_multipart_upload(bucket, object, opts)
                 .await
-                .map(|res| (res, 0));
+                .map(|res| (res, 0, opts.expected_bucket_incarnation_id));
+        }
+
+        if opts.data_movement && opts.version_id.is_some() {
+            let idx = self.select_data_movement_pool_idx(bucket, object, -1, opts, false).await?;
+            if idx == opts.src_pool_idx {
+                return Err(StorageError::DataMovementOverwriteErr(
+                    bucket.to_owned(),
+                    object.to_owned(),
+                    opts.version_id.clone().unwrap_or_default(),
+                ));
+            }
+            let res = self.pools[idx].new_multipart_upload(bucket, object, opts).await?;
+            return Ok((res, idx, opts.expected_bucket_incarnation_id));
         }
 
         for (idx, pool) in self.pools.iter().enumerate() {
@@ -372,7 +455,7 @@ impl ECStore {
 
             if !res.uploads.is_empty() {
                 let res = self.pools[idx].new_multipart_upload(bucket, object, opts).await?;
-                return Ok((res, idx));
+                return Ok((res, idx, opts.expected_bucket_incarnation_id));
             }
         }
         let idx = self.get_pool_idx(bucket, object, -1).await?;
@@ -385,7 +468,7 @@ impl ECStore {
         }
 
         let res = self.pools[idx].new_multipart_upload(bucket, object, opts).await?;
-        Ok((res, idx))
+        Ok((res, idx, opts.expected_bucket_incarnation_id))
     }
 
     #[instrument(skip(self))]
@@ -456,6 +539,30 @@ impl ECStore {
         Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
     }
 
+    pub(crate) async fn put_object_part_for_data_movement(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<PartInfo> {
+        let part_id = opts
+            .part_number
+            .ok_or_else(|| Error::other("targeted multipart upload requires a part number"))?;
+        check_put_object_part_args(bucket, object, upload_id)?;
+        if !opts.data_movement {
+            return Err(Error::other("targeted multipart upload requires data_movement options"));
+        }
+        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let pool = self
+            .pools
+            .get(target_pool_idx)
+            .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?;
+        pool.put_object_part(bucket, object, upload_id, part_id, data, &opts).await
+    }
+
     #[instrument(skip(self))]
     pub(super) async fn handle_get_multipart_info(
         &self,
@@ -530,6 +637,26 @@ impl ECStore {
         Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
     }
 
+    pub(crate) async fn abort_multipart_upload_for_data_movement(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        opts: &ObjectOptions,
+    ) -> Result<()> {
+        check_abort_multipart_args(bucket, object, upload_id)?;
+        if !opts.data_movement {
+            return Err(Error::other("targeted multipart abort requires data_movement options"));
+        }
+        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let pool = self
+            .pools
+            .get(target_pool_idx)
+            .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?;
+        pool.abort_multipart_upload(bucket, object, upload_id, &opts).await
+    }
+
     #[instrument(skip(self))]
     pub(super) async fn handle_complete_multipart_upload(
         self: Arc<Self>,
@@ -574,6 +701,62 @@ impl ECStore {
 
         Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
     }
+
+    pub(crate) async fn complete_multipart_upload_for_data_movement(
+        self: Arc<Self>,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        uploaded_parts: Vec<CompletePart>,
+        opts: &ObjectOptions,
+    ) -> Result<ObjectInfo> {
+        check_complete_multipart_args(bucket, object, upload_id)?;
+        if !opts.data_movement {
+            return Err(Error::other("targeted multipart completion requires data_movement options"));
+        }
+        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        if opts.overwrites_existing_version() && !is_meta_bucketname(bucket) {
+            let expected_incarnation_id = opts
+                .expected_bucket_incarnation_id
+                .ok_or_else(|| Error::other("data movement completion is missing its bucket incarnation"))?;
+            let lifecycle_fence = opts
+                .bucket_lifecycle_lock_fence
+                .as_ref()
+                .ok_or_else(|| Error::other("data movement completion is missing its bucket lifecycle fence"))?;
+            let snapshot = match opts.object_lock_config_snapshot.as_ref() {
+                Some(snapshot) => Arc::clone(snapshot),
+                None => {
+                    self.object_lock_config_snapshot_under_lifecycle_fence(bucket, lifecycle_fence)
+                        .await?
+                }
+            };
+            if !snapshot.is_valid_for_destructive_put(self.id, bucket, expected_incarnation_id) {
+                return Err(Error::other(
+                    "data movement Object Lock snapshot does not match the target bucket generation",
+                ));
+            }
+            snapshot.add_lock_fences(&mut opts);
+            opts.object_lock_config_snapshot = Some(snapshot);
+        }
+        #[cfg(test)]
+        pause_data_movement_multipart_before_selected_completion(bucket).await;
+        let pool = self
+            .pools
+            .get(target_pool_idx)
+            .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?
+            .clone();
+        let result = enqueue_transition_after_write(
+            pool.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, &opts)
+                .await,
+            LcEventSrc::S3CompleteMultipartUpload,
+        )
+        .await;
+        if result.is_ok() {
+            list_objects::observe_list_objects_mutation(self.as_ref(), bucket).await;
+        }
+        result
+    }
 }
 
 /// Merges per-pool `ListMultipartUploads` pages into a single globally paginated

crates/ecstore/src/store/object.rs

@@ -918,7 +918,7 @@ fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &Objec
         && is_data_movement_delete_marker(target)
         && source.version_id == target.version_id
         && source.mod_time == target.mod_time
-        && source.user_defined == target.user_defined
+        && is_equivalent_data_movement_delete_marker_metadata(&source.user_defined, &target.user_defined)
         && source.user_tags == target.user_tags
         && source.replication_status_internal == target.replication_status_internal
         && source.replication_status == target.replication_status
@@ -926,24 +926,185 @@ fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &Objec
         && source.version_purge_status == target.version_purge_status
 }
 
+fn is_equivalent_data_movement_delete_marker_metadata(
+    source: &HashMap<String, String>,
+    target: &HashMap<String, String>,
+) -> bool {
+    matches!(
+        (
+            data_movement_delete_marker_metadata_identity(source),
+            data_movement_delete_marker_metadata_identity(target)
+        ),
+        (Some(source), Some(target)) if source == target
+    )
+}
+
+fn data_movement_delete_marker_metadata_identity(metadata: &HashMap<String, String>) -> Option<HashMap<String, String>> {
+    let mut identity = HashMap::with_capacity(metadata.len());
+    let mut local_tier_free_version_id = None;
+    for (key, value) in metadata {
+        let Some(suffix) = rustfs_utils::http::strip_internal_prefix_preserving_case(key) else {
+            identity.insert(key.clone(), value.clone());
+            continue;
+        };
+
+        if suffix.eq_ignore_ascii_case(rustfs_utils::http::SUFFIX_TIER_FV_ID) {
+            let version_id = Uuid::parse_str(value).ok().filter(|version_id| !version_id.is_nil())?;
+            if local_tier_free_version_id.is_some_and(|expected| expected != version_id) {
+                return None;
+            }
+            local_tier_free_version_id = Some(version_id);
+            continue;
+        }
+
+        let canonical_suffix = [
+            rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+            rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+            rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            rustfs_utils::http::SUFFIX_PURGESTATUS,
+        ]
+        .into_iter()
+        .find(|candidate| suffix.eq_ignore_ascii_case(candidate))
+        .map(str::to_string)
+        .or_else(|| {
+            [
+                rustfs_utils::http::SUFFIX_REPLICATION_RESET_ARN_PREFIX,
+                rustfs_utils::http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX,
+            ]
+            .into_iter()
+            .find_map(|prefix| {
+                suffix
+                    .get(..prefix.len())
+                    .is_some_and(|candidate| candidate.eq_ignore_ascii_case(prefix))
+                    .then(|| format!("{prefix}{}", &suffix[prefix.len()..]))
+            })
+        })
+        .unwrap_or_else(|| suffix.to_string());
+        let canonical_value = if canonical_suffix.eq_ignore_ascii_case(rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP)
+            || canonical_suffix.eq_ignore_ascii_case(rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP)
+        {
+            rustfs_filemeta::parse_replication_timestamp(value)?
+                .unix_timestamp_nanos()
+                .to_string()
+        } else {
+            value.clone()
+        };
+        let canonical_key = format!("{}{canonical_suffix}", rustfs_utils::http::RUSTFS_INTERNAL_PREFIX);
+        if identity
+            .insert(canonical_key, canonical_value.clone())
+            .is_some_and(|existing| existing != canonical_value)
+        {
+            return None;
+        }
+    }
+    for (status_suffix, timestamp_suffix) in [
+        (rustfs_utils::http::SUFFIX_REPLICA_STATUS, rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP),
+        (
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+        ),
+    ] {
+        let status_key = format!("{}{status_suffix}", rustfs_utils::http::RUSTFS_INTERNAL_PREFIX);
+        let timestamp_key = format!("{}{timestamp_suffix}", rustfs_utils::http::RUSTFS_INTERNAL_PREFIX);
+        match (identity.contains_key(&status_key), identity.contains_key(&timestamp_key)) {
+            (true, false) => {
+                identity.insert(timestamp_key, OffsetDateTime::UNIX_EPOCH.unix_timestamp_nanos().to_string());
+            }
+            (false, true) => return None,
+            _ => {}
+        }
+    }
+    Some(identity)
+}
+
 fn is_data_movement_delete_marker(info: &ObjectInfo) -> bool {
     info.delete_marker
 }
 
+fn is_expected_data_movement_delete_marker_source(source: &ObjectInfo, expected_mod_time: Option<OffsetDateTime>) -> bool {
+    is_data_movement_delete_marker(source)
+        && source.mod_time.is_some()
+        && source.mod_time == expected_mod_time
+        && data_movement_delete_marker_metadata_identity(&source.user_defined).is_some()
+}
+
+fn current_data_movement_delete_marker_opts(source: &ObjectInfo, opts: &ObjectOptions) -> Option<ObjectOptions> {
+    let replica_status = rustfs_utils::http::get_str(&source.user_defined, rustfs_utils::http::SUFFIX_REPLICA_STATUS);
+    let replica_timestamp = rustfs_utils::http::get_str(&source.user_defined, rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP);
+    let (replica_status, replica_timestamp) = match (replica_status, replica_timestamp) {
+        (None, None) => Default::default(),
+        (Some(status), timestamp) => {
+            let status = crate::bucket::replication::ReplicationStatusType::from(status.as_str());
+            if status.is_empty() {
+                return None;
+            }
+            let timestamp = match timestamp {
+                Some(timestamp) => rustfs_filemeta::parse_replication_timestamp(&timestamp)?,
+                None => OffsetDateTime::UNIX_EPOCH,
+            };
+            (status, Some(timestamp))
+        }
+        (None, Some(_)) => return None,
+    };
+    let replication_status = rustfs_utils::http::get_str(&source.user_defined, rustfs_utils::http::SUFFIX_REPLICATION_STATUS);
+    let replication_timestamp =
+        rustfs_utils::http::get_str(&source.user_defined, rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP);
+    let (replication_status, replication_timestamp, replication_targets) = match (replication_status, replication_timestamp) {
+        (None, None) => Default::default(),
+        (Some(status), timestamp) => {
+            let direct_status = crate::bucket::replication::ReplicationStatusType::from(status.as_str());
+            let targets = crate::bucket::replication::replication_statuses_map(status.as_str());
+            if direct_status.is_empty() && targets.is_empty() {
+                return None;
+            }
+            let timestamp = match timestamp {
+                Some(timestamp) => rustfs_filemeta::parse_replication_timestamp(&timestamp)?,
+                None => OffsetDateTime::UNIX_EPOCH,
+            };
+            (Some(status), Some(timestamp), targets)
+        }
+        (None, Some(_)) => return None,
+    };
+    let mut state = source.replication_state();
+    if state.target_delete_marker_version_ids_corrupt {
+        return None;
+    }
+    state.replica_status = replica_status;
+    state.replica_timestamp = replica_timestamp;
+    state.replication_status_internal = replication_status;
+    state.replication_timestamp = replication_timestamp;
+    state.targets = replication_targets;
+    state.replicate_decision_str = source.replication_decision.clone();
+    state.delete_marker = true;
+
+    let mut target_opts = opts.clone();
+    target_opts.mod_time = source.mod_time;
+    target_opts.delete_replication = Some(state);
+    Some(target_opts)
+}
+
 fn expected_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo) -> ObjectInfo {
     ObjectInfo::from_file_info(source, "", &source.name, source.version_id.is_some())
 }
 
 fn is_equivalent_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo, target: &ObjectInfo) -> bool {
     let expected = expected_data_movement_tiered_object(source);
+    let Some(source_actual_size) = effective_object_actual_size(&expected) else {
+        return false;
+    };
+    let Some(target_actual_size) = effective_object_actual_size(target) else {
+        return false;
+    };
 
     source.version_id == target.version_id
         && !target.delete_marker
         && source.size == target.size
         && source.get_etag() == target.etag
         && source.checksum == target.checksum
+        && crate::data_movement::are_equivalent_data_movement_parts(&source.parts, &target.parts)
         && source.mod_time == target.mod_time
-        && expected.user_defined == target.user_defined
+        && crate::data_movement::is_equivalent_data_movement_metadata(&expected, target, source_actual_size, target_actual_size)
         && expected.user_tags == target.user_tags
         && expected.expires == target.expires
         && expected.storage_class == target.storage_class
@@ -952,11 +1113,12 @@ fn is_equivalent_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo,
         && expected.version_purge_status_internal == target.version_purge_status_internal
         && expected.version_purge_status == target.version_purge_status
         && expected.transitioned_object.status == target.transitioned_object.status
+        && expected.transition_version_state == target.transition_version_state
         && expected.transitioned_object.name == target.transitioned_object.name
         && expected.transitioned_object.tier == target.transitioned_object.tier
         && expected.transitioned_object.version_id == target.transitioned_object.version_id
         && expected.transitioned_object.free_version == target.transitioned_object.free_version
-        && effective_object_actual_size(target) == Some(source.size)
+        && source_actual_size == target_actual_size
 }
 
 fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool {
@@ -1054,7 +1216,7 @@ impl ECStore {
         )))
     }
 
-    async fn object_lock_config_snapshot_under_lifecycle_fence(
+    pub(super) async fn object_lock_config_snapshot_under_lifecycle_fence(
         &self,
         bucket: &str,
         lifecycle_fence: &NamespaceLockFence,
@@ -1454,7 +1616,8 @@ impl ECStore {
         target_pool_idx: usize,
         opts: &ObjectOptions,
     ) -> Result<Option<ObjectInfo>> {
-        let lookup_opts = version_aware_lookup_opts(opts, true);
+        let mut lookup_opts = version_aware_lookup_opts(opts, true);
+        lookup_opts.include_part_checksums = true;
 
         let Some(pool) = self.pools.get(target_pool_idx) else {
             return Err(Error::other(format!(
@@ -1521,6 +1684,25 @@ impl ECStore {
     ) -> Result<()> {
         check_put_object_args(bucket, object)?;
 
+        let mut opts = opts.clone();
+        let bucket_incarnation_fence = if is_meta_bucketname(bucket) {
+            None
+        } else {
+            let expected = opts
+                .expected_bucket_incarnation_id
+                .ok_or_else(|| Error::other("tiered data movement is missing its bucket incarnation snapshot"))?;
+            let guard = self.acquire_bucket_incarnation_fence(bucket, expected).await?;
+            if let Some(namespace_guard) = guard.namespace_lock_guard() {
+                opts.add_bucket_lifecycle_lock_guard(namespace_guard);
+            }
+            Some(guard)
+        };
+
+        let mut fi = fi.clone();
+        if opts.data_movement {
+            crate::data_movement::prepare_tiered_data_movement_file_info(&mut fi)?;
+        }
+
         let object = encode_dir_object(object);
 
         if self.single_pool() {
@@ -1533,7 +1715,8 @@ impl ECStore {
 
         let idx = if opts.data_movement && opts.version_id.is_some() {
             Self::resolve_decommission_target_pool_idx_result(
-                self.select_data_movement_pool_idx(bucket, &object, fi.size, opts, true).await,
+                self.select_data_movement_pool_idx(bucket, &object, fi.size, &opts, true)
+                    .await,
                 bucket,
                 &object,
             )?
@@ -1550,7 +1733,7 @@ impl ECStore {
                 .await;
             let target_pool_idx = resolve_data_movement_resume_target_pool(idx, resume_target_pool_idx, opts.src_pool_idx);
             if self
-                .has_equivalent_data_movement_tiered_object(bucket, &object, fi, opts, target_pool_idx)
+                .has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, target_pool_idx)
                 .await?
             {
                 return Ok(());
@@ -1563,14 +1746,27 @@ impl ECStore {
             ));
         }
 
-        Self::resolve_decommission_tiered_object_result(
-            self.pools[idx]
-                .get_disks_by_key(&object)
-                .decommission_tiered_object(bucket, &object, fi, opts)
-                .await,
-            bucket,
-            &object,
-        )
+        let result = self.pools[idx]
+            .get_disks_by_key(&object)
+            .decommission_tiered_object(bucket, &object, &fi, &opts)
+            .await;
+        if matches!(result, Err(Error::PreconditionFailed)) {
+            if self
+                .has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, idx)
+                .await?
+            {
+                return Ok(());
+            }
+            return Err(StorageError::DataMovementOverwriteErr(
+                bucket.to_owned(),
+                object,
+                opts.version_id.clone().unwrap_or_default(),
+            ));
+        }
+        if bucket_incarnation_fence.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+            return Err(Error::other("tiered data movement bucket incarnation fence was lost during target write"));
+        }
+        Self::resolve_decommission_tiered_object_result(result, bucket, &object)
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -1607,15 +1803,7 @@ impl ECStore {
         Ok(Self::attach_read_lock_guard(reader, read_lock_guard))
     }
 
-    #[instrument(level = "debug", skip(self, data))]
-    #[hotpath::measure(impl_type = "ECStore")]
-    pub(super) async fn handle_put_object(
-        &self,
-        bucket: &str,
-        object: &str,
-        data: &mut PutObjReader,
-        opts: &ObjectOptions,
-    ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
+    async fn prepare_put_object(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<(String, ObjectOptions)> {
         check_put_object_args(bucket, object)?;
 
         let object = encode_dir_object(object);
@@ -1642,22 +1830,20 @@ impl ECStore {
             };
             snapshot.add_lock_fences(&mut opts);
         }
+        Ok((object, opts))
+    }
 
-        // Keep PUT atomic-read friendly: SetDisks takes the object write lock only
-        // around precondition checks and the final rename/commit.
+    async fn select_put_object_pool_idx(&self, bucket: &str, object: &str, size: i64, opts: &ObjectOptions) -> Result<usize> {
         if self.single_pool() {
-            return self.pools[0]
-                .put_object_with_old_current_size(bucket, object.as_str(), data, &opts)
-                .await;
+            return Ok(0);
         }
 
         let idx = if opts.data_movement && opts.version_id.is_some() {
-            self.select_data_movement_pool_idx(bucket, &object, data.size(), &opts, false)
-                .await?
+            self.select_data_movement_pool_idx(bucket, object, size, opts, false).await?
         } else if opts.no_lock {
-            self.get_pool_idx_no_lock(bucket, &object, data.size()).await?
+            self.get_pool_idx_no_lock(bucket, object, size).await?
         } else {
-            self.get_pool_idx(bucket, &object, data.size()).await?
+            self.get_pool_idx(bucket, object, size).await?
         };
 
         if opts.data_movement && idx == opts.src_pool_idx {
@@ -1667,7 +1853,50 @@ impl ECStore {
                 opts.version_id.clone().unwrap_or_default(),
             ));
         }
+        Ok(idx)
+    }
 
+    pub(crate) async fn put_object_for_data_movement(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<(usize, Result<ObjectInfo>)> {
+        if !opts.data_movement {
+            return Err(Error::other("data movement PUT requires data_movement options"));
+        }
+        let (object, opts) = self.prepare_put_object(bucket, object, opts).await?;
+        let idx = self
+            .select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
+            .await?;
+        let result = self.pools[idx]
+            .put_object_with_old_current_size(bucket, &object, data, &opts)
+            .await
+            .map(|(object_info, _)| object_info);
+        let result = enqueue_transition_after_write(result, LcEventSrc::S3PutObject).await;
+        if result.is_ok() {
+            list_objects::observe_list_objects_mutation(self, bucket).await;
+        }
+        Ok((idx, result))
+    }
+
+    #[instrument(level = "debug", skip(self, data))]
+    #[hotpath::measure(impl_type = "ECStore")]
+    pub(super) async fn handle_put_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
+        let (object, opts) = self.prepare_put_object(bucket, object, opts).await?;
+        let idx = self
+            .select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
+            .await?;
+
+        // Keep PUT atomic-read friendly: SetDisks takes the object write lock only
+        // around precondition checks and the final rename/commit.
         self.pools[idx]
             .put_object_with_old_current_size(bucket, &object, data, &opts)
             .await
@@ -2110,6 +2339,50 @@ impl ECStore {
             };
             let target_pool_idx =
                 resolve_data_movement_resume_target_pool(selected_target_pool_idx, resume_target_pool_idx, opts.src_pool_idx);
+            let mut delete_marker_target_opts = None;
+
+            if opts.delete_marker && should_check_data_movement_resume_target(opts.src_pool_idx, target_pool_idx) {
+                let source = self
+                    .find_data_movement_target_info(bucket, object, opts.src_pool_idx, &opts)
+                    .await?;
+                let Some(source) = source else {
+                    return Err(StorageError::DataMovementOverwriteErr(
+                        bucket.to_owned(),
+                        object.to_owned(),
+                        opts.version_id.unwrap_or_default(),
+                    ));
+                };
+                if !is_expected_data_movement_delete_marker_source(&source, opts.mod_time) {
+                    return Err(StorageError::DataMovementOverwriteErr(
+                        bucket.to_owned(),
+                        object.to_owned(),
+                        opts.version_id.unwrap_or_default(),
+                    ));
+                }
+                let Some(target_opts) = current_data_movement_delete_marker_opts(&source, &opts) else {
+                    return Err(StorageError::DataMovementOverwriteErr(
+                        bucket.to_owned(),
+                        object.to_owned(),
+                        opts.version_id.unwrap_or_default(),
+                    ));
+                };
+                let target = self
+                    .find_data_movement_target_info(bucket, object, target_pool_idx, &target_opts)
+                    .await?;
+                if let Some(target) = target {
+                    if is_equivalent_data_movement_delete_marker(&source, &target) {
+                        let mut target = target;
+                        target.name = decode_dir_object(object);
+                        return Ok(target);
+                    }
+                    return Err(StorageError::DataMovementOverwriteErr(
+                        bucket.to_owned(),
+                        object.to_owned(),
+                        opts.version_id.unwrap_or_default(),
+                    ));
+                }
+                delete_marker_target_opts = Some(target_opts);
+            }
 
             if !should_check_data_movement_resume_target(opts.src_pool_idx, target_pool_idx) {
                 if let Ok((source_pool_info, _)) = existing_pool_info
@@ -2137,7 +2410,8 @@ impl ECStore {
                 ));
             }
 
-            let mut obj = self.pools[target_pool_idx].delete_object(bucket, object, opts).await?;
+            let target_opts = delete_marker_target_opts.unwrap_or(opts);
+            let mut obj = self.pools[target_pool_idx].delete_object(bucket, object, target_opts).await?;
             obj.name = decode_dir_object(obj.name.as_str());
             return Ok(obj);
         }
@@ -3200,6 +3474,247 @@ mod tests {
         assert!(!is_equivalent_data_movement_delete_marker(&source, &mismatched));
     }
 
+    #[test]
+    fn equivalent_data_movement_delete_marker_accepts_distinct_local_free_version_ids() {
+        let mut source = ObjectInfo {
+            version_id: Some(Uuid::from_u128(1)),
+            delete_marker: true,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            ..Default::default()
+        };
+        rustfs_utils::http::insert_str(
+            Arc::make_mut(&mut source.user_defined),
+            rustfs_utils::http::SUFFIX_TIER_FV_ID,
+            Uuid::from_u128(2).to_string(),
+        );
+        let mut target = source.clone();
+        rustfs_utils::http::insert_str(
+            Arc::make_mut(&mut target.user_defined),
+            rustfs_utils::http::SUFFIX_TIER_FV_ID,
+            Uuid::from_u128(3).to_string(),
+        );
+
+        assert!(is_equivalent_data_movement_delete_marker(&source, &target));
+
+        Arc::make_mut(&mut target.user_defined).insert(
+            format!("{}{}", rustfs_utils::http::MINIO_INTERNAL_PREFIX, rustfs_utils::http::SUFFIX_TIER_FV_ID),
+            Uuid::from_u128(4).to_string(),
+        );
+        assert!(!is_equivalent_data_movement_delete_marker(&source, &target));
+    }
+
+    #[test]
+    fn equivalent_data_movement_delete_marker_accepts_replication_alias_expansion() {
+        let key = format!(
+            "{}{}",
+            rustfs_utils::http::MINIO_INTERNAL_PREFIX,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS
+        );
+        let timestamp_key = format!(
+            "{}{}",
+            rustfs_utils::http::MINIO_INTERNAL_PREFIX,
+            rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP
+        );
+        let source = ObjectInfo {
+            version_id: Some(Uuid::from_u128(1)),
+            delete_marker: true,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            user_defined: Arc::new(HashMap::from([
+                (key.clone(), "arn=COMPLETED;".to_string()),
+                (timestamp_key, "1970-01-01T00:00:01Z".to_string()),
+            ])),
+            ..Default::default()
+        };
+        let mut target = source.clone();
+        rustfs_utils::http::insert_str(
+            Arc::make_mut(&mut target.user_defined),
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            "arn=COMPLETED;".to_string(),
+        );
+        rustfs_utils::http::insert_str(
+            Arc::make_mut(&mut target.user_defined),
+            rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+            (OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND).to_string(),
+        );
+        assert!(is_equivalent_data_movement_delete_marker(&source, &target));
+
+        Arc::make_mut(&mut target.user_defined).insert(key, "arn=FAILED;".to_string());
+        assert!(!is_equivalent_data_movement_delete_marker(&source, &target));
+    }
+
+    #[test]
+    fn data_movement_delete_marker_source_requires_persisted_mod_time() {
+        let source = ObjectInfo {
+            delete_marker: true,
+            ..Default::default()
+        };
+        assert!(!is_expected_data_movement_delete_marker_source(&source, None));
+
+        let source = ObjectInfo {
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            ..source
+        };
+        assert!(is_expected_data_movement_delete_marker_source(&source, Some(OffsetDateTime::UNIX_EPOCH)));
+        assert!(!is_expected_data_movement_delete_marker_source(&source, None));
+    }
+
+    #[test]
+    fn data_movement_delete_marker_uses_current_source_replication_state() {
+        let expected_timestamp = OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND;
+        let timestamp = expected_timestamp.to_string();
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+            ReplicationStatusType::Replica.to_string(),
+        );
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP, timestamp.clone());
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP, timestamp);
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            "arn=COMPLETED;".to_string(),
+        );
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            &format!(
+                "{}{}",
+                rustfs_utils::http::SUFFIX_REPLICATION_RESET_ARN_PREFIX,
+                "arn:minio:replication::TenantA:bucket"
+            ),
+            "reset-id".to_string(),
+        );
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            &format!(
+                "{}{}",
+                rustfs_utils::http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX,
+                "arn:minio:replication::TenantA:bucket"
+            ),
+            "target-version".to_string(),
+        );
+        let source = ObjectInfo {
+            delete_marker: true,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            replication_status_internal: Some("arn=COMPLETED;".to_string()),
+            replication_decision: "arn=replicate;".to_string(),
+            user_defined: Arc::new(metadata),
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            mod_time: source.mod_time,
+            delete_replication: Some(ReplicationState {
+                replication_status_internal: Some("arn=PENDING;".to_string()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+
+        let target_opts = current_data_movement_delete_marker_opts(&source, &opts).expect("valid current source state");
+        let state = target_opts.delete_replication.as_ref().expect("current replication state");
+        assert_eq!(state.replication_status_internal.as_deref(), Some("arn=COMPLETED;"));
+        assert_eq!(state.replica_status, crate::bucket::replication::ReplicationStatusType::Replica);
+        assert_eq!(state.replica_timestamp, Some(expected_timestamp));
+        assert_eq!(state.replication_timestamp, state.replica_timestamp);
+        assert_eq!(state.replicate_decision_str, "arn=replicate;");
+        assert_eq!(
+            state
+                .reset_statuses_map
+                .get("arn:minio:replication::TenantA:bucket")
+                .map(String::as_str),
+            Some("reset-id")
+        );
+        assert_eq!(
+            state
+                .target_delete_marker_version_ids
+                .get("arn:minio:replication::TenantA:bucket")
+                .map(String::as_str),
+            Some("target-version")
+        );
+    }
+
+    #[test]
+    fn data_movement_delete_marker_rejects_corrupt_target_version_maps() {
+        let suffix = format!("{}not-an-arn", rustfs_utils::http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX);
+        let mut malformed = HashMap::new();
+        rustfs_utils::http::insert_str(&mut malformed, &suffix, "target-version".to_string());
+        let malformed_source = ObjectInfo {
+            delete_marker: true,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            user_defined: Arc::new(malformed),
+            ..Default::default()
+        };
+        assert!(current_data_movement_delete_marker_opts(&malformed_source, &ObjectOptions::default()).is_none());
+
+        let mut conflicted = HashMap::new();
+        let suffix = format!(
+            "{}arn:minio:replication::target:bucket",
+            rustfs_utils::http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX
+        );
+        rustfs_utils::http::insert_str(&mut conflicted, &suffix, "target-version-a".to_string());
+        conflicted.insert(
+            format!("{}{suffix}", rustfs_utils::http::MINIO_INTERNAL_PREFIX),
+            "target-version-b".to_string(),
+        );
+        let conflicted_source = ObjectInfo {
+            user_defined: Arc::new(conflicted),
+            ..malformed_source.clone()
+        };
+        assert!(current_data_movement_delete_marker_opts(&conflicted_source, &ObjectOptions::default()).is_none());
+
+        let mut over_cap = HashMap::new();
+        for index in 0..=1_000 {
+            let suffix = format!(
+                "{}arn:minio:replication::target:bucket-{index}",
+                rustfs_utils::http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX
+            );
+            rustfs_utils::http::insert_str(&mut over_cap, &suffix, format!("target-version-{index}"));
+        }
+        let over_cap_source = ObjectInfo {
+            user_defined: Arc::new(over_cap),
+            ..malformed_source
+        };
+        assert!(current_data_movement_delete_marker_opts(&over_cap_source, &ObjectOptions::default()).is_none());
+    }
+
+    #[test]
+    fn data_movement_delete_marker_normalizes_legacy_missing_replication_timestamps() {
+        let mut source_metadata = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut source_metadata,
+            rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+            ReplicationStatusType::Replica.to_string(),
+        );
+        rustfs_utils::http::insert_str(
+            &mut source_metadata,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            "arn=COMPLETED;".to_string(),
+        );
+        let source = ObjectInfo {
+            delete_marker: true,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            replication_status_internal: Some("arn=COMPLETED;".to_string()),
+            user_defined: Arc::new(source_metadata),
+            ..Default::default()
+        };
+
+        let target_opts = current_data_movement_delete_marker_opts(&source, &ObjectOptions::default())
+            .expect("legacy status-only metadata should remain migratable");
+        let state = target_opts
+            .delete_replication
+            .expect("replication state should be reconstructed");
+        assert_eq!(state.replica_timestamp, Some(OffsetDateTime::UNIX_EPOCH));
+        assert_eq!(state.replication_timestamp, Some(OffsetDateTime::UNIX_EPOCH));
+
+        let mut target_metadata = (*source.user_defined).clone();
+        let epoch = OffsetDateTime::UNIX_EPOCH
+            .format(&time::format_description::well_known::Rfc3339)
+            .unwrap();
+        rustfs_utils::http::insert_str(&mut target_metadata, rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP, epoch.clone());
+        rustfs_utils::http::insert_str(&mut target_metadata, rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP, epoch);
+        assert!(is_equivalent_data_movement_delete_marker_metadata(&source.user_defined, &target_metadata));
+    }
+
     #[test]
     fn equivalent_data_movement_delete_marker_rejects_metadata_and_replication_mismatch() {
         let version_id = Uuid::nil();
@@ -3344,6 +3859,67 @@ mod tests {
         assert!(is_equivalent_data_movement_tiered_object(&source, &target));
     }
 
+    #[test]
+    fn equivalent_data_movement_tiered_object_uses_logical_compressed_and_encrypted_sizes() {
+        let mut compressed = tiered_equivalence_source();
+        compressed.size = 600;
+        rustfs_utils::http::insert_str(&mut compressed.metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string());
+        rustfs_utils::http::insert_str(&mut compressed.metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "1024".to_string());
+        let compressed_target = tiered_equivalence_target(&compressed);
+        assert!(is_equivalent_data_movement_tiered_object(&compressed, &compressed_target));
+
+        let mut encrypted = tiered_equivalence_source();
+        encrypted.size = 640;
+        encrypted.metadata.insert(
+            rustfs_utils::http::object_encryption_keys::INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(),
+            "key-id".to_string(),
+        );
+        encrypted.metadata.insert(
+            rustfs_utils::http::object_encryption_keys::INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER.to_string(),
+            "1024".to_string(),
+        );
+        let encrypted_target = tiered_equivalence_target(&encrypted);
+        assert!(is_equivalent_data_movement_tiered_object(&encrypted, &encrypted_target));
+    }
+
+    #[test]
+    fn equivalent_data_movement_tiered_object_accepts_transition_alias_expansion() {
+        let mut source = tiered_equivalence_source();
+        let suffix = rustfs_utils::http::SUFFIX_TRANSITION_TIER;
+        source.metadata.insert(
+            format!("{}{suffix}", rustfs_utils::http::MINIO_INTERNAL_PREFIX),
+            source.transition_tier.clone(),
+        );
+        let mut target = tiered_equivalence_target(&source);
+        Arc::make_mut(&mut target.user_defined)
+            .insert(rustfs_utils::http::internal_key_rustfs(suffix), source.transition_tier.clone());
+
+        assert!(is_equivalent_data_movement_tiered_object(&source, &target));
+    }
+
+    #[test]
+    fn equivalent_data_movement_tiered_object_requires_hydrated_part_checksums() {
+        let mut source = tiered_equivalence_source();
+        source.parts = vec![rustfs_filemeta::ObjectPartInfo {
+            number: 1,
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND),
+            checksums: Some(HashMap::from([("CRC32C".to_string(), "AAAAAA==".to_string())])),
+            ..Default::default()
+        }];
+        rustfs_utils::http::insert_str(
+            &mut source.metadata,
+            rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
+            r#"[[1,[["CRC32C","AAAAAA=="]]]]"#.to_string(),
+        );
+        let mut target = tiered_equivalence_target(&source);
+        Arc::make_mut(&mut target.parts)[0].mod_time = None;
+        assert!(is_equivalent_data_movement_tiered_object(&source, &target));
+
+        let mut missing = target;
+        Arc::make_mut(&mut missing.parts)[0].checksums = None;
+        assert!(!is_equivalent_data_movement_tiered_object(&source, &missing));
+    }
+
     #[test]
     fn equivalent_data_movement_tiered_object_rejects_transition_mismatch() {
         let source = tiered_equivalence_source();
@@ -3353,6 +3929,16 @@ mod tests {
         assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
     }
 
+    #[test]
+    fn equivalent_data_movement_tiered_object_rejects_transition_version_state_mismatch() {
+        let mut source = tiered_equivalence_source();
+        source.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact;
+        let mut target = tiered_equivalence_target(&source);
+        target.transition_version_state = rustfs_filemeta::TransitionVersionState::Unknown;
+
+        assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
+    }
+
     #[test]
     fn equivalent_data_movement_tiered_object_rejects_user_metadata_mismatch() {
         let source = tiered_equivalence_source();