Merge remote-tracking branch 'origin/main' into feat/sse-ecstore-decoupling

# Conflicts:
#	crates/kms/src/service_manager.rs

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -47,8 +47,8 @@ use crate::diagnostics::get::{
     record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
 use crate::disk::{
-    OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction,
-    part_transaction_path,
+    DataDirDeleteStatus, OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK,
+    PartTransactionAction, part_transaction_path,
 };
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::ShardReader;
@@ -547,6 +547,8 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
         && left.transitioned_objname == right.transitioned_objname
         && left.transition_tier == right.transition_tier
         && left.transition_version_id == right.transition_version_id
+        && left.transition_version == right.transition_version
+        && left.transition_version_state == right.transition_version_state
         && left.expire_restored == right.expire_restored
         && left.size == right.size
         && left.mod_time == right.mod_time
@@ -2985,29 +2987,48 @@ impl SetDisks {
                 Self::rename_fanout_barrier(&object_for_fault, idx, rename_fanout_barrier_phase::CLEANUP).await;
 
                 if let Some(err) = Self::cleanup_injected_error(&object_for_fault, idx) {
-                    return Some(err);
+                    return (false, Some(err));
                 }
                 if let Some(disk) = disk {
-                    disk.delete(
-                        &bucket,
-                        &file_path,
-                        DeleteOptions {
-                            recursive: true,
-                            ..Default::default()
-                        },
-                    )
-                    .await
-                    .err()
+                    match disk
+                        .delete_data_dir(
+                            &bucket,
+                            &file_path,
+                            DeleteOptions {
+                                recursive: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                    {
+                        Ok(DataDirDeleteStatus::Deleted) => (false, None),
+                        Ok(DataDirDeleteStatus::Deferred) => (true, None),
+                        Err(err) => (false, Some(err)),
+                    }
                 } else {
                     // `None` slot: ignored placeholder. It is not `attempted`, so
                     // classification excludes it from residue regardless.
-                    Some(DiskError::DiskNotFound)
+                    (false, Some(DiskError::DiskNotFound))
                 }
             })
         });
-        let errs: Vec<Option<DiskError>> = join_all(futures).await.into_iter().map(map_cleanup_join_result).collect();
+        let mut deferred = 0usize;
+        let errs: Vec<Option<DiskError>> = join_all(futures)
+            .await
+            .into_iter()
+            .map(|result| match result {
+                Ok((was_deferred, err)) => {
+                    deferred += usize::from(was_deferred);
+                    err
+                }
+                Err(join_err) => Some(DiskError::other(format!("old data dir cleanup task failed: {join_err}"))),
+            })
+            .collect();
 
-        classify_old_data_dir_cleanup(&errs, &attempted, write_quorum)
+        let mut cleanup = classify_old_data_dir_cleanup(&errs, &attempted, write_quorum);
+        cleanup.deferred = deferred;
+        cleanup.reclaimed = cleanup.reclaimed.saturating_sub(deferred);
+        cleanup
     }
 
     /// Test-only fault-injection seam for the old-data-dir cleanup path
@@ -3098,6 +3119,20 @@ impl SetDisks {
 
         rustfs_io_metrics::record_old_data_dir_cleanup(c.attempted, c.reclaimed, c.unreclaimed_disks.len(), c.below_quorum);
 
+        if c.deferred > 0 {
+            debug!(
+                event = EVENT_SET_DISK_WRITE,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_SET_DISK,
+                bucket = %bucket,
+                object = %object,
+                old_data_dir = %old_dir,
+                deferred = c.deferred,
+                state = "old_data_cleanup_deferred",
+                "Old data directory cleanup deferred for active snapshot leases"
+            );
+        }
+
         if actions.warn {
             warn!(
                 component = LOG_COMPONENT_ECSTORE,
@@ -4129,6 +4164,9 @@ pub(in crate::set_disk) struct OldDataDirCleanup {
     /// Number of attempted disks that returned `Ok` or a not-found variant
     /// (a missing dir == already reclaimed).
     pub reclaimed: usize,
+    /// Number of attempted disks that retained the directory for an active
+    /// snapshot lease and registered it for deletion after the final release.
+    pub deferred: usize,
     /// Indices of attempted disks that failed with a non-ignored, non-not-found
     /// error (including task panic/cancel). This is the residue that actually
     /// leaks and drives the leak metric + heal enqueue.
@@ -4191,6 +4229,7 @@ fn classify_old_data_dir_cleanup(errs: &[Option<DiskError>], attempted: &[bool],
     OldDataDirCleanup {
         attempted: attempted_count,
         reclaimed,
+        deferred: 0,
         unreclaimed_disks,
         below_quorum,
     }
@@ -5131,6 +5170,40 @@ mod tests {
         drop((disk1, disk2));
     }
 
+    #[tokio::test]
+    async fn commit_cleanup_reports_and_releases_deferred_snapshot_data_dirs() {
+        let bucket = "cleanup-lease-bucket";
+        let object = "cleanup-lease-object";
+        let old_data_dir = "11111111-1111-1111-1111-111111111111";
+        let committed_data_dir = "22222222-2222-2222-2222-222222222222";
+        let data_dir_path = format!("{object}/{old_data_dir}");
+        let shard_path = format!("{data_dir_path}/part.1");
+        let (_dir1, disk1) = read_multiple_test_disk(bucket, &[(&shard_path, b"one".as_slice())]).await;
+        let set = io_primitives_test_set(vec![Some(disk1.clone())], 0).await;
+        let lease = disk1
+            .acquire_snapshot_lease(bucket, &data_dir_path)
+            .await
+            .expect("snapshot lease should be acquired before cleanup");
+
+        let cleanup = set
+            .commit_rename_data_dir(&[Some(disk1.clone())], bucket, object, old_data_dir, committed_data_dir, 1)
+            .await;
+        assert_eq!(cleanup.attempted, 1);
+        assert_eq!(cleanup.reclaimed, 0);
+        assert_eq!(cleanup.deferred, 1);
+        assert!(cleanup.unreclaimed_disks.is_empty());
+        disk1
+            .read_all(bucket, &shard_path)
+            .await
+            .expect("deferred cleanup must leave later shard opens available");
+
+        disk1
+            .release_snapshot_lease(bucket, &data_dir_path, lease)
+            .await
+            .expect("final lease release should reclaim the old data directory");
+        assert!(matches!(disk1.read_all(bucket, &shard_path).await, Err(DiskError::FileNotFound)));
+    }
+
     /// Isolation guard: an armed barrier / observed object only affects its own
     /// object. A fan-out for a different (unobserved, unarmed) object must not be
     /// paused and must not accrue any tracked task count — so concurrent tests

crates/ecstore/src/set_disk/metadata.rs

@@ -553,7 +553,7 @@ impl SetDisks {
         }
     }
 
-    fn file_info_quorum_hash(meta: &FileInfo) -> [u8; 32] {
+    pub(super) fn file_info_quorum_hash(meta: &FileInfo) -> [u8; 32] {
         let mut hasher = Sha256::new();
         Self::update_file_info_quorum_hash(&mut hasher, meta);
         let digest = hasher.finalize();
@@ -576,6 +576,13 @@ impl SetDisks {
         Self::update_hash_str(hasher, &meta.transition_tier);
         Self::update_hash_str(hasher, &meta.transitioned_objname);
         Self::update_hash_optional_uuid(hasher, meta.transition_version_id);
+        Self::update_hash_optional_str(hasher, meta.transition_version.as_deref());
+        hasher.update([match meta.transition_version_state {
+            rustfs_filemeta::TransitionVersionState::Unknown => 0,
+            rustfs_filemeta::TransitionVersionState::KnownDisabled => 1,
+            rustfs_filemeta::TransitionVersionState::SuspendedNull => 2,
+            rustfs_filemeta::TransitionVersionState::Exact => 3,
+        }]);
         Self::update_hash_optional_u32(hasher, meta.mode);
         Self::update_hash_optional_u64(hasher, meta.written_by_version);
 

crates/ecstore/src/set_disk/mod.rs

@@ -686,9 +686,13 @@ mod core;
 mod ctx;
 mod metadata;
 mod ops;
+#[cfg(test)]
+pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
+#[cfg(test)]
+pub(crate) use ops::object::cleanup_rejected_transition_upload_durably;
 mod read;
 mod replication;
 pub(crate) mod shard_source;

crates/ecstore/src/set_disk/ops/heal.rs

@@ -92,6 +92,69 @@ struct PartFailureSummary {
     bitrot_failure: bool,
 }
 
+#[derive(Clone)]
+struct RecoverableMetaCandidate {
+    identity: [u8; 32],
+    file_info: FileInfo,
+    data_count: usize,
+    local_payload: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum DanglingDeleteSafety {
+    UnsafeToDelete,
+    NoRecoverableCandidate,
+}
+
+#[cfg(test)]
+struct DanglingCheckPartsFailure {
+    key: DanglingCheckPartsFailureKey,
+}
+
+#[cfg(test)]
+type DanglingCheckPartsFailureKey = (String, String, usize);
+
+#[cfg(test)]
+type DanglingCheckPartsFailures = HashMap<DanglingCheckPartsFailureKey, DiskError>;
+
+#[cfg(test)]
+fn dangling_check_parts_failures() -> &'static std::sync::Mutex<DanglingCheckPartsFailures> {
+    static FAILURES: std::sync::OnceLock<std::sync::Mutex<DanglingCheckPartsFailures>> = std::sync::OnceLock::new();
+    FAILURES.get_or_init(|| std::sync::Mutex::new(HashMap::new()))
+}
+
+#[cfg(test)]
+impl DanglingCheckPartsFailure {
+    fn install(bucket: &str, object: &str, disk_index: usize, error: DiskError) -> Self {
+        let key = (bucket.to_string(), object.to_string(), disk_index);
+        let previous = dangling_check_parts_failures()
+            .lock()
+            .expect("dangling check-parts failure registry should not poison")
+            .insert(key.clone(), error);
+        assert!(previous.is_none(), "dangling check-parts failure already installed");
+        Self { key }
+    }
+}
+
+#[cfg(test)]
+impl Drop for DanglingCheckPartsFailure {
+    fn drop(&mut self) {
+        dangling_check_parts_failures()
+            .lock()
+            .expect("dangling check-parts failure registry should not poison")
+            .remove(&self.key);
+    }
+}
+
+#[cfg(test)]
+fn injected_dangling_check_parts_error(bucket: &str, object: &str, disk_index: usize) -> Option<DiskError> {
+    dangling_check_parts_failures()
+        .lock()
+        .expect("dangling check-parts failure registry should not poison")
+        .get(&(bucket.to_string(), object.to_string(), disk_index))
+        .cloned()
+}
+
 fn first_unhealthy_part_summary(
     data_errs_by_part: &HashMap<usize, Vec<usize>>,
     parts: &[ObjectPartInfo],
@@ -125,39 +188,17 @@ impl SetDisks {
         version_id: &str,
         opts: &HealOpts,
     ) -> disk::error::Result<(HealResultItem, Option<DiskError>)> {
-        // `allow_meta_regen` is true on the first pass: a version whose data shards
-        // physically survive (>= data_blocks) but whose xl.meta fell below
-        // read-quorum is RESCUED (missing xl.meta regenerated) rather than
-        // dangling-deleted. The re-drive after a rescue sets it false so the
-        // regeneration can happen at most once (no unbounded recursion).
-        Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, true)).await
-    }
-
-    /// Best-effort orphan-data-dir reclaim for an object that is healthy on this
-    /// set. Wraps [`Self::reclaim_orphan_data_dirs`] with the shared logging so
-    /// both `heal_object` exits — the already-healthy early return and the
-    /// post-heal tail — reclaim identically. Never fails the heal: delete errors
-    /// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
-    async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
-        match self.reclaim_orphan_data_dirs(bucket, object).await {
-            Ok(removed) if removed > 0 => {
-                info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
-            }
-            Ok(_) => {}
-            Err(e) => {
-                warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
-            }
-        }
+        Box::pin(self.heal_object_with_explicit_version_regen(bucket, object, version_id, opts, true)).await
     }
 
     #[allow(clippy::too_many_lines)]
-    async fn heal_object_with_regen(
+    async fn heal_object_with_explicit_version_regen(
         &self,
         bucket: &str,
         object: &str,
         version_id: &str,
         opts: &HealOpts,
-        allow_meta_regen: bool,
+        allow_explicit_version_regen: bool,
     ) -> disk::error::Result<(HealResultItem, Option<DiskError>)> {
         info!(?opts, "Starting heal_object");
 
@@ -354,22 +395,6 @@ impl SetDisks {
                             }
                         }
 
-                        // DATA-SAFETY GUARD (backlog#920, decision 1): before any
-                        // dangling delete, if the version's DATA shards physically
-                        // survive on >= data_blocks disks it is RECONSTRUCTABLE.
-                        // Regenerate the missing xl.meta from a surviving valid
-                        // FileInfo and re-drive the heal instead of destroying a
-                        // recoverable version. Torn writes (< data_blocks data
-                        // shards) fall through to the existing dangling behavior.
-                        if cannot_heal
-                            && allow_meta_regen
-                            && self
-                                .try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks)
-                                .await?
-                        {
-                            return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await;
-                        }
-
                         if cannot_heal {
                             let total_disks = parts_metadata.len();
                             let healthy_count = total_disks.saturating_sub(disks_to_heal_count);
@@ -422,6 +447,20 @@ impl SetDisks {
                                 );
                             }
 
+                            // `disks_with_all_parts` normalizes conflicting entries
+                            // in `parts_metadata` to defaults. Re-read only before
+                            // destructive cleanup so the guard sees every original
+                            // identity.
+                            let (delete_guard_metadata, delete_guard_errs) =
+                                Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
+                            if self
+                                .dangling_delete_safety(bucket, object, &delete_guard_metadata, &delete_guard_errs, &disks)
+                                .await?
+                                == DanglingDeleteSafety::UnsafeToDelete
+                            {
+                                return Ok((result, Some(cannot_heal_err)));
+                            }
+
                             // Allow for dangling deletes, on versions that have DataDir missing etc.
                             // this would end up restoring the correct readable versions.
                             return match self
@@ -837,17 +876,25 @@ impl SetDisks {
                 }
             }
             Err(err) => {
-                // DATA-SAFETY GUARD (backlog#920, decision 1): meta quorum failed,
-                // but the version's DATA may still physically survive on enough
-                // disks (xl.meta lost on > parity disks while part files remain).
-                // Rescue it by regenerating the missing xl.meta and re-driving heal
-                // instead of dangling-deleting a reconstructable version.
-                if allow_meta_regen
+                if allow_explicit_version_regen
+                    && !version_id.is_empty()
                     && self
-                        .try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks)
+                        .try_regenerate_explicit_version_meta(bucket, object, version_id, &parts_metadata, &errs, &disks)
                         .await?
                 {
-                    return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await;
+                    return Box::pin(self.heal_object_with_explicit_version_regen(bucket, object, version_id, opts, false)).await;
+                }
+
+                if self
+                    .dangling_delete_safety(bucket, object, &parts_metadata, &errs, &disks)
+                    .await?
+                    == DanglingDeleteSafety::UnsafeToDelete
+                {
+                    return Ok((
+                        self.default_heal_result(FileInfo::default(), &errs, bucket, object, version_id)
+                            .await,
+                        Some(err),
+                    ));
                 }
 
                 let data_errs_by_part = HashMap::new();
@@ -883,129 +930,226 @@ impl SetDisks {
         }
     }
 
-    /// backlog#920 (decision 1): rescue a version that meta-quorum logic would
-    /// otherwise dangling-DELETE, when its DATA is still reconstructable.
-    ///
-    /// Returns `Ok(true)` if the version was rescued (missing xl.meta regenerated
-    /// on at least one disk, so a re-driven heal can reconstruct it), `Ok(false)`
-    /// to fall through to the existing dangling-delete behavior.
-    ///
-    /// Recoverability is computed by physically probing part files across ALL
-    /// disks in the set with `check_parts` — including disks whose xl.meta is
-    /// absent (a lost xl.meta does not lose the sibling `part.*` data). If at
-    /// least `data_blocks` disks hold every part of a surviving valid FileInfo,
-    /// the object is EC-reconstructable, so we regenerate that FileInfo's xl.meta
-    /// on every disk whose metadata is absent (via `write_metadata`, which merges
-    /// into any existing xl.meta). Delete markers, remote/transitioned versions,
-    /// and genuine torn writes (< `data_blocks` surviving data shards) are NOT
-    /// rescued — they keep the current dangling-delete-after-grace behavior, so no
-    /// regression on those paths.
-    async fn try_regenerate_recoverable_meta(
+    async fn try_regenerate_explicit_version_meta(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        disks: &[Option<DiskStore>],
+    ) -> disk::error::Result<bool> {
+        let Ok(version_id) = Uuid::parse_str(version_id) else {
+            return Ok(false);
+        };
+        let candidates = parts_metadata
+            .iter()
+            .zip(errs.iter())
+            .filter_map(|(file_info, err)| {
+                (err.is_none()
+                    && file_info_is_valid_for_metadata(file_info)
+                    && file_info.version_id == Some(version_id)
+                    && file_info.has_valid_erasure_geometry()
+                    && !file_info.deleted
+                    && !file_info.is_remote()
+                    && file_info.data_dir.is_some()
+                    && !file_info.parts.is_empty()
+                    && file_info.erasure.data_blocks > 0
+                    && file_info
+                        .erasure
+                        .data_blocks
+                        .checked_add(file_info.erasure.parity_blocks)
+                        .is_some_and(|shards| shards == disks.len()))
+                .then_some(file_info)
+            })
+            .collect::<Vec<_>>();
+        let Some(candidate) = candidates.first().copied() else {
+            return Ok(false);
+        };
+        let identity = Self::file_info_quorum_hash(candidate);
+        if candidates
+            .iter()
+            .any(|file_info| Self::file_info_quorum_hash(file_info) != identity)
+        {
+            return Ok(false);
+        }
+
+        let mut available = 0usize;
+        for disk in disks {
+            let Some(disk) = disk else {
+                return Ok(false);
+            };
+            match disk.check_parts(bucket, object, candidate).await {
+                Ok(response)
+                    if !response.results.is_empty() && response.results.iter().all(|result| *result == CHECK_PART_SUCCESS) =>
+                {
+                    available += 1;
+                }
+                Ok(_)
+                | Err(
+                    DiskError::FileNotFound
+                    | DiskError::FileVersionNotFound
+                    | DiskError::PathNotFound
+                    | DiskError::VolumeNotFound,
+                ) => {}
+                Err(_) => return Ok(false),
+            }
+        }
+        if available < candidate.erasure.data_blocks {
+            return Ok(false);
+        }
+
+        let mut wrote = 0usize;
+        for (index, disk) in disks.iter().enumerate() {
+            let Some(disk) = disk else {
+                return Ok(false);
+            };
+            let metadata_absent = matches!(
+                errs.get(index).and_then(Option::as_ref),
+                Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)
+            );
+            if !metadata_absent {
+                continue;
+            }
+            let Some(&shard_index) = candidate.erasure.distribution.get(index) else {
+                return Ok(false);
+            };
+            let mut regenerated = candidate.clone();
+            regenerated.fresh = false;
+            regenerated.erasure.index = shard_index;
+            match disk.write_metadata("", bucket, object, regenerated).await {
+                Ok(()) => wrote += 1,
+                Err(error) => {
+                    warn!(
+                        bucket,
+                        object,
+                        disk_index = index,
+                        error = %error,
+                        "failed to regenerate recoverable xl.meta"
+                    );
+                }
+            }
+        }
+        Ok(wrote > 0)
+    }
+
+    /// Best-effort orphan-data-dir reclaim for an object that is healthy on this
+    /// set. Wraps [`Self::reclaim_orphan_data_dirs`] with the shared logging so
+    /// both `heal_object` exits — the already-healthy early return and the
+    /// post-heal tail — reclaim identically. Never fails the heal: delete errors
+    /// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
+    async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
+        match self.reclaim_orphan_data_dirs(bucket, object).await {
+            Ok(removed) if removed > 0 => {
+                info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
+            }
+            Ok(_) => {}
+            Err(e) => {
+                warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
+            }
+        }
+    }
+
+    /// Prevent dangling cleanup when surviving state cannot prove that deletion
+    /// is safe. Part presence proves only recoverability, never commit: the write
+    /// path can durably rename data before xl.meta is committed.
+    async fn dangling_delete_safety(
         &self,
         bucket: &str,
         object: &str,
         parts_metadata: &[FileInfo],
         errs: &[Option<DiskError>],
         disks: &[Option<DiskStore>],
-    ) -> disk::error::Result<bool> {
-        // A surviving valid, non-deleted, non-remote data FileInfo to rebuild from.
-        let Some(surviving) = parts_metadata
+    ) -> disk::error::Result<DanglingDeleteSafety> {
+        if disks.iter().any(Option::is_none)
+            || errs.iter().flatten().any(|err| {
+                !matches!(
+                    err,
+                    DiskError::FileNotFound
+                        | DiskError::FileVersionNotFound
+                        | DiskError::PathNotFound
+                        | DiskError::VolumeNotFound
+                )
+            })
+        {
+            return Ok(DanglingDeleteSafety::UnsafeToDelete);
+        }
+
+        let mut candidates = Vec::<RecoverableMetaCandidate>::with_capacity(parts_metadata.len());
+        for (fi, err) in parts_metadata.iter().zip(errs.iter()) {
+            if err.is_some() || !file_info_is_valid_for_metadata(fi) {
+                continue;
+            }
+
+            let identity = Self::file_info_quorum_hash(fi);
+            if !candidates.iter().any(|candidate| candidate.identity == identity) {
+                let local_payload = fi.has_valid_erasure_geometry()
+                    && !fi.deleted
+                    && !fi.is_remote()
+                    && fi.data_dir.is_some()
+                    && !fi.parts.is_empty()
+                    && fi.erasure.data_blocks > 0
+                    && fi
+                        .erasure
+                        .data_blocks
+                        .checked_add(fi.erasure.parity_blocks)
+                        .is_some_and(|shards| shards == disks.len());
+                candidates.push(RecoverableMetaCandidate {
+                    identity,
+                    file_info: fi.clone(),
+                    data_count: 0,
+                    local_payload,
+                });
+            }
+        }
+
+        if candidates
             .iter()
-            .find(|fi| fi.has_valid_erasure_geometry() && !fi.deleted && !fi.is_remote())
-            .cloned()
-        else {
-            return Ok(false);
-        };
-
-        // Without a data_dir + parts there is no data to prove recoverable.
-        if surviving.data_dir.is_none() || surviving.parts.is_empty() {
-            return Ok(false);
-        }
-        let data_blocks = surviving.erasure.data_blocks;
-        if data_blocks == 0 {
-            return Ok(false);
+            .any(|candidate| candidate.file_info.deleted || candidate.file_info.is_remote())
+            || candidates.len() > 1
+        {
+            return Ok(DanglingDeleteSafety::UnsafeToDelete);
         }
 
-        // Physically probe part presence on EVERY online disk using the surviving
-        // FileInfo's data_dir/parts. `check_parts` stats `object/<data_dir>/part.N`
-        // directly, so it counts disks that still hold the data even if their
-        // xl.meta was deleted.
-        let mut available = 0usize;
-        for disk in disks.iter().flatten() {
-            if let Ok(resp) = disk.check_parts(bucket, object, &surviving).await
-                && !resp.results.is_empty()
-                && resp.results.iter().all(|r| *r == CHECK_PART_SUCCESS)
-            {
-                available += 1;
-            }
-        }
+        for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
+            for (disk_index, disk) in disks.iter().enumerate() {
+                let Some(disk) = disk else {
+                    return Ok(DanglingDeleteSafety::UnsafeToDelete);
+                };
+                #[cfg(test)]
+                let check_result = match injected_dangling_check_parts_error(bucket, object, disk_index) {
+                    Some(error) => Err(error),
+                    None => disk.check_parts(bucket, object, &candidate.file_info).await,
+                };
+                #[cfg(not(test))]
+                let check_result = disk.check_parts(bucket, object, &candidate.file_info).await;
 
-        // Torn write: fewer than data_blocks surviving data shards is genuinely
-        // unrecoverable — preserve the current dangling behavior (no resurrection).
-        if available < data_blocks {
-            debug!(
-                bucket,
-                object,
-                available,
-                data_blocks,
-                "heal_object: version not reconstructable (torn write), keeping dangling behavior"
-            );
-            return Ok(false);
-        }
-
-        // Reconstructable: regenerate the surviving xl.meta on every disk whose
-        // metadata is absent so the version regains read-quorum. Each disk gets its
-        // OWN shard index: the disk at physical position `index` holds shard
-        // `distribution[index]` (mirrors `shuffle_disks` + the write path's
-        // `erasure.index = shuffled_pos + 1`). Copying the surviving disk's index
-        // verbatim would write an inconsistent xl.meta that the re-heal then treats
-        // as corrupt.
-        let distribution = &surviving.erasure.distribution;
-        let mut wrote = 0usize;
-        for (index, disk) in disks.iter().enumerate() {
-            let Some(disk) = disk else { continue };
-            let meta_absent = matches!(
-                errs.get(index).and_then(Option::as_ref),
-                Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)
-            ) || !parts_metadata.get(index).map(FileInfo::is_valid).unwrap_or(false);
-            if !meta_absent {
-                continue;
-            }
-            // Without a known shard index for this position we cannot write a
-            // consistent xl.meta; leave it for the normal heal to reconstruct.
-            let Some(&shard_index) = distribution.get(index) else {
-                continue;
-            };
-            let mut regen = surviving.clone();
-            regen.fresh = false; // merge into any existing xl.meta on the disk
-            regen.erasure.index = shard_index;
-            match disk.write_metadata("", bucket, object, regen).await {
-                Ok(()) => wrote += 1,
-                Err(e) => {
-                    warn!(
-                        bucket,
-                        object,
-                        disk_index = index,
-                        error = %e,
-                        "heal_object: failed to regenerate recoverable xl.meta on disk"
-                    );
+                match check_result {
+                    Ok(resp) if !resp.results.is_empty() && resp.results.iter().all(|result| *result == CHECK_PART_SUCCESS) => {
+                        candidate.data_count += 1;
+                    }
+                    Ok(_) => {}
+                    Err(
+                        DiskError::FileNotFound
+                        | DiskError::FileVersionNotFound
+                        | DiskError::PathNotFound
+                        | DiskError::VolumeNotFound,
+                    ) => {}
+                    Err(_) => return Ok(DanglingDeleteSafety::UnsafeToDelete),
                 }
             }
         }
 
-        if wrote == 0 {
-            return Ok(false);
-        }
-
-        info!(
-            bucket,
-            object,
-            available,
-            data_blocks,
-            regenerated_meta_disks = wrote,
-            "heal_object: rescued reconstructable sub-quorum version by regenerating xl.meta"
-        );
-        Ok(true)
+        Ok(
+            if candidates
+                .iter()
+                .any(|candidate| candidate.local_payload && candidate.data_count >= candidate.file_info.erasure.data_blocks)
+            {
+                DanglingDeleteSafety::UnsafeToDelete
+            } else {
+                DanglingDeleteSafety::NoRecoverableCandidate
+            },
+        )
     }
 
     pub(in crate::set_disk) async fn heal_object_dir_locked(
@@ -1393,7 +1537,7 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
 
 #[cfg(test)]
 mod heal_result_report_tests {
-    use super::SetDisks;
+    use super::{DanglingCheckPartsFailure, DanglingDeleteSafety, SetDisks};
     use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope};
     use crate::disk::endpoint::Endpoint;
     use crate::disk::error::DiskError;
@@ -1405,10 +1549,12 @@ mod heal_result_report_tests {
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
     use crate::{config::storageclass, store::init_format::save_format_file};
     use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode};
-    use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo};
+    use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE};
     use std::sync::Arc;
     use tempfile::TempDir;
+    use time::OffsetDateTime;
     use tokio::sync::RwLock;
+    use uuid::Uuid;
 
     async fn real_disk() -> (TempDir, Endpoint, DiskStore) {
         let dir = tempfile::tempdir().expect("tempdir should be created");
@@ -1446,6 +1592,68 @@ mod heal_result_report_tests {
         .await
     }
 
+    fn meta_regen_test_fileinfo(object: &str, data_dir: Uuid, mod_time: i64, disk_index: usize) -> FileInfo {
+        let mut fi = FileInfo::new(object, 2, 2);
+        fi.data_dir = Some(data_dir);
+        fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(mod_time).expect("test timestamp should parse"));
+        fi.size = 1;
+        fi.parts = vec![ObjectPartInfo {
+            number: 1,
+            size: 1,
+            actual_size: 1,
+            ..Default::default()
+        }];
+        fi.erasure.index = fi.erasure.distribution[disk_index];
+        fi
+    }
+
+    async fn meta_regen_test_set(
+        bucket: &str,
+        object: &str,
+        data_dirs: &[(Uuid, usize)],
+    ) -> (Vec<TempDir>, Arc<SetDisks>, Vec<Option<DiskStore>>) {
+        let mut temp_dirs = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+        for disk_index in 0..4 {
+            let (temp_dir, endpoint, disk) = real_disk().await;
+            disk.make_volume(bucket).await.expect("test bucket should be created");
+            for (data_dir, shard_count) in data_dirs {
+                if disk_index >= *shard_count {
+                    continue;
+                }
+                let part_dir = temp_dir.path().join(bucket).join(object).join(data_dir.to_string());
+                tokio::fs::create_dir_all(&part_dir)
+                    .await
+                    .expect("test data directory should be created");
+                tokio::fs::write(part_dir.join("part.1"), [1u8; 2])
+                    .await
+                    .expect("test data shard should be written");
+            }
+            temp_dirs.push(temp_dir);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+
+        let set = set_disks_with(disks.clone(), endpoints, 2).await;
+        (temp_dirs, set, disks)
+    }
+
+    async fn seed_meta_regen_test_metadata(
+        disks: &[Option<DiskStore>],
+        disk_index: usize,
+        bucket: &str,
+        object: &str,
+        file_info: &FileInfo,
+    ) {
+        disks[disk_index]
+            .as_ref()
+            .expect("metadata test disk should be online")
+            .write_metadata("", bucket, object, file_info.clone())
+            .await
+            .expect("test metadata should be written");
+    }
+
     async fn formatted_single_disk_no_parity_set() -> (TempDir, Arc<SetDisks>) {
         let format = FormatV3::new(1, 1);
         let dir = tempfile::tempdir().expect("tempdir should be created");
@@ -1753,6 +1961,312 @@ mod heal_result_report_tests {
         assert_eq!(result.before.drives[3].state, DriveState::Ok.to_string());
     }
 
+    #[tokio::test]
+    async fn dangling_delete_guard_preserves_conflicting_identities_without_writing_metadata() {
+        let bucket = "bucket-delete-guard-conflict";
+        let object = "object.bin";
+        let old_data_dir = Uuid::parse_str("99999999-9999-9999-9999-999999999999").expect("old data dir should parse");
+        let new_data_dir = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").expect("new data dir should parse");
+        let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(old_data_dir, 4), (new_data_dir, 2)]).await;
+        let version_id = Uuid::parse_str("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb").expect("version id should parse");
+        let mut metadata = vec![
+            meta_regen_test_fileinfo(object, old_data_dir, 9, 0),
+            meta_regen_test_fileinfo(object, new_data_dir, 10, 1),
+            FileInfo::default(),
+            FileInfo::default(),
+        ];
+        metadata[0].version_id = Some(version_id);
+        metadata[1].version_id = Some(version_id);
+        assert_eq!(
+            metadata[0].version_id, metadata[1].version_id,
+            "the conflicting candidates must share one version id"
+        );
+        seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata[0]).await;
+        seed_meta_regen_test_metadata(&disks, 1, bucket, object, &metadata[1]).await;
+        let errs = vec![None, None, Some(DiskError::FileNotFound), Some(DiskError::FileNotFound)];
+
+        assert!(
+            set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
+                .await
+                .expect("conflicting identities should be classified")
+                == DanglingDeleteSafety::UnsafeToDelete
+        );
+        let reversed = vec![
+            metadata[1].clone(),
+            metadata[0].clone(),
+            FileInfo::default(),
+            FileInfo::default(),
+        ];
+        assert!(
+            set.dangling_delete_safety(bucket, object, &reversed, &errs, &disks)
+                .await
+                .expect("reversed identities should be classified")
+                == DanglingDeleteSafety::UnsafeToDelete
+        );
+        let version_id = version_id.to_string();
+        assert!(
+            !set.try_regenerate_explicit_version_meta(bucket, object, &version_id, &metadata, &errs, &disks)
+                .await
+                .expect("conflicting explicit-version candidates should be rejected"),
+            "an explicit version must not select between conflicting metadata identities"
+        );
+        for disk_index in [2, 3] {
+            assert!(
+                matches!(
+                    disks[disk_index]
+                        .as_ref()
+                        .expect("test disk should be online")
+                        .read_version("", bucket, object, "", &ReadOptions::default())
+                        .await,
+                    Err(DiskError::FileNotFound)
+                ),
+                "the delete guard must not manufacture metadata on missing disks"
+            );
+        }
+        let old = disks[0]
+            .as_ref()
+            .expect("first test disk should be online")
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("old metadata should remain readable");
+        let new = disks[1]
+            .as_ref()
+            .expect("second test disk should be online")
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("new metadata should remain readable");
+        assert_eq!(old.data_dir, Some(old_data_dir));
+        assert_eq!(new.data_dir, Some(new_data_dir));
+    }
+
+    #[tokio::test]
+    async fn heal_meta_quorum_failure_preserves_reconstructable_uncommitted_candidate() {
+        let bucket = "bucket-delete-guard-reconstructable";
+        let object = "object.bin";
+        let data_dir = Uuid::parse_str("33333333-3333-3333-3333-333333333333").expect("data dir should parse");
+        let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
+        let metadata = [
+            meta_regen_test_fileinfo(object, data_dir, 3, 0),
+            FileInfo::default(),
+            FileInfo::default(),
+            FileInfo::default(),
+        ];
+        seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata[0]).await;
+        let (observed_metadata, observed_errs) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", true, true, false)
+            .await
+            .expect("test metadata should be readable across the set");
+        assert_eq!(
+            set.dangling_delete_safety(bucket, object, &observed_metadata, &observed_errs, &disks)
+                .await
+                .expect("observed reconstructable candidate should be classified"),
+            DanglingDeleteSafety::UnsafeToDelete
+        );
+
+        let (_, err) = set
+            .heal_object(
+                bucket,
+                object,
+                "",
+                &HealOpts {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("unsafe dangling state should be reported without deletion");
+        assert_eq!(err, Some(DiskError::FileNotFound));
+        let surviving = disks[0]
+            .as_ref()
+            .expect("first test disk should be online")
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("the only metadata copy must be preserved");
+        assert_eq!(surviving.data_dir, Some(data_dir));
+        assert!(
+            matches!(
+                disks[1]
+                    .as_ref()
+                    .expect("second test disk should be online")
+                    .read_version("", bucket, object, "", &ReadOptions::default())
+                    .await,
+                Err(DiskError::FileNotFound)
+            ),
+            "the delete guard must not propagate metadata"
+        );
+    }
+
+    #[tokio::test]
+    async fn heal_meta_quorum_failure_preserves_candidate_when_required_shard_disk_is_offline() {
+        let bucket = "bucket-delete-guard-offline";
+        let object = "object.bin";
+        let data_dir = Uuid::parse_str("44444444-4444-4444-4444-444444444444").expect("data dir should parse");
+        let (temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
+        let metadata = meta_regen_test_fileinfo(object, data_dir, 4, 0);
+        seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata).await;
+        set.disks.write().await[1] = None;
+
+        let (_, err) = set
+            .heal_object(
+                bucket,
+                object,
+                "",
+                &HealOpts {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("offline shard state should be reported without deletion");
+        assert_eq!(err, Some(DiskError::FileNotFound));
+        let surviving = disks[0]
+            .as_ref()
+            .expect("first test disk should be online")
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("offline uncertainty must preserve the surviving metadata");
+        assert_eq!(surviving.data_dir, Some(data_dir));
+        assert!(
+            temp_dirs[0]
+                .path()
+                .join(bucket)
+                .join(object)
+                .join(data_dir.to_string())
+                .join("part.1")
+                .is_file(),
+            "offline uncertainty must preserve the last online shard"
+        );
+    }
+
+    #[tokio::test]
+    async fn heal_meta_quorum_failure_preserves_candidate_when_part_probe_times_out() {
+        let bucket = "bucket-delete-guard-timeout";
+        let object = "object.bin";
+        let data_dir = Uuid::parse_str("55555555-5555-5555-5555-555555555555").expect("data dir should parse");
+        let (temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
+        let metadata = meta_regen_test_fileinfo(object, data_dir, 5, 0);
+        seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata).await;
+        let _failure = DanglingCheckPartsFailure::install(bucket, object, 1, DiskError::Timeout);
+
+        let (_, err) = set
+            .heal_object(
+                bucket,
+                object,
+                "",
+                &HealOpts {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("part probe timeout should be reported without deletion");
+        assert_eq!(err, Some(DiskError::FileNotFound));
+        let surviving = disks[0]
+            .as_ref()
+            .expect("first test disk should be online")
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("probe uncertainty must preserve the surviving metadata");
+        assert_eq!(surviving.data_dir, Some(data_dir));
+        assert!(
+            temp_dirs[0]
+                .path()
+                .join(bucket)
+                .join(object)
+                .join(data_dir.to_string())
+                .join("part.1")
+                .is_file(),
+            "probe uncertainty must preserve the last confirmed shard"
+        );
+    }
+
+    #[tokio::test]
+    async fn dangling_delete_guard_ignores_set_incompatible_geometry() {
+        let bucket = "bucket-delete-guard-short-geometry";
+        let object = "object.bin";
+        let data_dir = Uuid::parse_str("abababab-abab-abab-abab-abababababab").expect("data dir should parse");
+        let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 1)]).await;
+        let mut candidate = FileInfo::new(object, 1, 0);
+        candidate.data_dir = Some(data_dir);
+        candidate.mod_time = Some(OffsetDateTime::from_unix_timestamp(18).expect("timestamp should parse"));
+        candidate.size = 1;
+        candidate.parts = vec![ObjectPartInfo {
+            number: 1,
+            size: 1,
+            actual_size: 1,
+            ..Default::default()
+        }];
+        candidate.erasure.index = candidate.erasure.distribution[0];
+        seed_meta_regen_test_metadata(&disks, 0, bucket, object, &candidate).await;
+        let metadata = vec![candidate, FileInfo::default(), FileInfo::default(), FileInfo::default()];
+        let errs = vec![
+            None,
+            Some(DiskError::FileNotFound),
+            Some(DiskError::FileNotFound),
+            Some(DiskError::FileNotFound),
+        ];
+
+        assert!(
+            set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
+                .await
+                .expect("set-incompatible geometry should be classified")
+                == DanglingDeleteSafety::NoRecoverableCandidate
+        );
+    }
+
+    #[tokio::test]
+    async fn dangling_delete_guard_preserves_delete_marker_and_remote_metadata() {
+        let bucket = "bucket-delete-guard-nonlocal";
+        let object = "object.bin";
+        let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[]).await;
+        let marker = FileInfo {
+            name: object.to_string(),
+            version_id: Some(Uuid::parse_str("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee").expect("version id should parse")),
+            deleted: true,
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(14).expect("marker timestamp should parse")),
+            ..Default::default()
+        };
+        let remote_dir = Uuid::parse_str("89898989-8989-8989-8989-898989898989").expect("remote data dir should parse");
+        let mut remote = meta_regen_test_fileinfo(object, remote_dir, 15, 1);
+        remote.transition_status = TRANSITION_COMPLETE.to_string();
+        remote.transition_tier = "WARM".to_string();
+        remote.transitioned_objname = "remote/object.bin".to_string();
+
+        for metadata in [marker, remote] {
+            let candidates = vec![metadata, FileInfo::default(), FileInfo::default(), FileInfo::default()];
+            let errs = vec![
+                None,
+                Some(DiskError::FileNotFound),
+                Some(DiskError::FileNotFound),
+                Some(DiskError::FileNotFound),
+            ];
+            assert_eq!(
+                set.dangling_delete_safety(bucket, object, &candidates, &errs, &disks)
+                    .await
+                    .expect("non-local metadata should be classified"),
+                DanglingDeleteSafety::UnsafeToDelete
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn dangling_delete_guard_preserves_metadata_read_uncertainty() {
+        let bucket = "bucket-delete-guard-read-error";
+        let object = "object.bin";
+        let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[]).await;
+        let metadata = vec![FileInfo::default(); disks.len()];
+
+        for read_error in [DiskError::Timeout, DiskError::DiskAccessDenied, DiskError::DiskNotFound] {
+            let mut errs = vec![Some(DiskError::FileNotFound); disks.len()];
+            errs[0] = Some(read_error);
+            assert_eq!(
+                set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
+                    .await
+                    .expect("metadata read uncertainty should be classified"),
+                DanglingDeleteSafety::UnsafeToDelete
+            );
+        }
+    }
+
     #[tokio::test]
     async fn heal_no_parity_bitrot_reports_unrecoverable_integrity_failure() {
         let (dir, set) = formatted_single_disk_no_parity_set().await;

crates/ecstore/src/set_disk/ops/list.rs

@@ -29,6 +29,12 @@ impl SetDisks {
     pub async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
         ListOperations::new(self.ctx()).delete_all(bucket, prefix).await
     }
+
+    pub(crate) async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
+        ListOperations::new(self.ctx())
+            .delete_all_with_quorum(bucket, prefix, write_quorum)
+            .await
+    }
 }
 
 /// List/prefix maintenance operations, borrowing the `SetDisks` core state
@@ -48,6 +54,14 @@ impl<'a> ListOperations<'a> {
     }
 
     pub(crate) async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
+        self.delete_all_inner(bucket, prefix, None).await
+    }
+
+    async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
+        self.delete_all_inner(bucket, prefix, Some(write_quorum)).await
+    }
+
+    async fn delete_all_inner(&self, bucket: &str, prefix: &str, write_quorum: Option<usize>) -> Result<()> {
         let disks = self.ctx.disks().read().await;
 
         let disks = disks.clone();
@@ -79,6 +93,9 @@ impl<'a> ListOperations<'a> {
                 Ok(_) => {
                     errors.push(None);
                 }
+                Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => {
+                    errors.push(None);
+                }
                 Err(e) => {
                     errors.push(Some(e));
                 }
@@ -97,6 +114,12 @@ impl<'a> ListOperations<'a> {
             );
         }
 
+        if let Some(write_quorum) = write_quorum
+            && let Some(err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, write_quorum)
+        {
+            return Err(err.into());
+        }
+
         Ok(())
     }
 }

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -26,6 +26,8 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
+#[cfg(test)]
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 use tokio::task::JoinSet;
 
@@ -33,7 +35,8 @@ const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
 #[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-enum MultipartCommitPause {
+pub(crate) enum MultipartCommitPause {
+    PutPartBeforeLockAcquire,
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
@@ -45,12 +48,14 @@ struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
     pause: MultipartCommitPause,
+    expected_arrivals: usize,
+    arrivals: AtomicUsize,
     arrived: tokio::sync::Notify,
-    release: tokio::sync::Notify,
+    release: tokio::sync::Semaphore,
 }
 
 #[cfg(test)]
-struct MultipartCommitBarrier {
+pub(crate) struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
@@ -60,13 +65,25 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc
 
 #[cfg(test)]
 impl MultipartCommitBarrier {
-    fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+        Self::install_for_arrivals(bucket, object, pause, 1)
+    }
+
+    pub(crate) fn install_for_arrivals(
+        bucket: &str,
+        object: &str,
+        pause: MultipartCommitPause,
+        expected_arrivals: usize,
+    ) -> Self {
+        assert!(expected_arrivals > 0, "multipart commit barrier must wait for at least one arrival");
         let state = Arc::new(MultipartCommitBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
             pause,
+            expected_arrivals,
+            arrivals: AtomicUsize::new(0),
             arrived: tokio::sync::Notify::new(),
-            release: tokio::sync::Notify::new(),
+            release: tokio::sync::Semaphore::new(0),
         });
         let mut slot = MULTIPART_COMMIT_BARRIER
             .get_or_init(|| std::sync::Mutex::new(None))
@@ -78,21 +95,29 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    async fn wait_until_paused(&self) {
-        tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
-            .await
-            .expect("multipart completion should reach the deterministic commit barrier");
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                let arrived = self.state.arrived.notified();
+                if self.state.arrivals.load(Ordering::Acquire) >= self.state.expected_arrivals {
+                    return;
+                }
+                arrived.await;
+            }
+        })
+        .await
+        .expect("multipart completion should reach the deterministic commit barrier");
     }
 
-    fn release(&self) {
-        self.state.release.notify_one();
+    pub(crate) fn release(&self) {
+        self.state.release.add_permits(self.state.expected_arrivals);
     }
 }
 
 #[cfg(test)]
 impl Drop for MultipartCommitBarrier {
     fn drop(&mut self) {
-        self.state.release.notify_one();
+        self.release();
         let mut slot = MULTIPART_COMMIT_BARRIER
             .get_or_init(|| std::sync::Mutex::new(None))
             .lock()
@@ -112,9 +137,21 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm
         .as_ref()
         .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
         .cloned();
-    if let Some(barrier) = barrier {
-        barrier.arrived.notify_one();
-        barrier.release.notified().await;
+    if let Some(barrier) = barrier
+        && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
+            (current < barrier.expected_arrivals).then_some(current + 1)
+        })
+    {
+        let arrival = previous + 1;
+        if arrival == barrier.expected_arrivals {
+            barrier.arrived.notify_one();
+        }
+        barrier
+            .release
+            .acquire()
+            .await
+            .expect("multipart commit barrier should remain open")
+            .forget();
     }
 }
 
@@ -687,6 +724,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
             let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix);
 
+            #[cfg(test)]
+            pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockAcquire).await;
             // Serialize only the commit (rename_part), not the whole upload. Each
             // concurrent stream writes to its own unique temp dir (see `tmp_part`
             // above), so the encode/stream phase never conflicts and must stay
@@ -777,6 +816,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         mut max_parts: usize,
         opts: &ObjectOptions,
     ) -> Result<ListPartsInfo> {
+        let _upload_guard = self
+            .acquire_multipart_upload_read_lock("list_object_parts", bucket, object, upload_id, opts)
+            .await?;
         let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, false).await?;
 
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
@@ -1254,10 +1296,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let _upload_guard = self
             .acquire_multipart_upload_write_lock("abort_multipart_upload", bucket, object, upload_id, opts)
             .await?;
-        self.check_upload_id_exists(bucket, object, upload_id, false).await?;
+        let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
-        self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await
+        self.delete_all_with_quorum(
+            RUSTFS_META_MULTIPART_BUCKET,
+            &upload_id_path,
+            fi.write_quorum(self.default_write_quorum()),
+        )
+        .await
     }
     // complete_multipart_upload finished
     #[tracing::instrument(skip(self))]
@@ -1815,7 +1862,36 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         #[cfg(test)]
         pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-        drop(upload_guard);
+
+        let cleanup_store = self.clone();
+        let cleanup_upload_id_path = upload_id_path.clone();
+        let cleanup_bucket = bucket.to_owned();
+        let cleanup_object = object.to_owned();
+        let cleanup_upload_id = upload_id.to_owned();
+        let cleanup_handle = tokio::spawn(async move {
+            let _upload_guard = upload_guard;
+            if let Err(err) = cleanup_store
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
+                .await
+            {
+                warn!(
+                    bucket = %cleanup_bucket,
+                    object = %cleanup_object,
+                    upload_id = %cleanup_upload_id,
+                    error = ?err,
+                    "completed multipart upload staging cleanup did not reach write quorum"
+                );
+            }
+        });
+        if let Err(err) = cleanup_handle.await {
+            warn!(
+                bucket = %bucket,
+                object = %object,
+                upload_id = %upload_id,
+                error = ?err,
+                "completed multipart upload staging cleanup task failed"
+            );
+        }
         drop(object_lock_guard); // drop object lock guard to release the lock
 
         // backlog#1321: enqueue heal only when the committed replicas actually
@@ -1860,12 +1936,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             });
         }
 
-        let upload_id_path = upload_id_path.clone();
-        let store = self.clone();
-        let _cleanup_handle = tokio::spawn(async move {
-            let _ = store.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await;
-        });
-
         for (i, op_disk) in online_disks.iter().enumerate() {
             if let Some(disk) = op_disk
                 && disk.is_online().await
@@ -2177,6 +2247,122 @@ mod tests {
         )
     }
 
+    async fn assert_complete_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x47; 4096], &create_opts).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                .await
+        });
+        barrier.wait_until_paused().await;
+
+        let abort_store = set_disks.clone();
+        let abort_upload_id = upload_id.clone();
+        let abort = tokio::spawn(async move {
+            abort_store
+                .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+        assert!(!abort.is_finished(), "abort must wait for the completion upload lock");
+
+        barrier.release();
+        complete
+            .await
+            .expect("completion task should not panic")
+            .expect("completion should win the upload finalization");
+        let abort_err = abort
+            .await
+            .expect("abort task should not panic")
+            .expect_err("abort must observe the upload as finalized");
+        assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
+        set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("complete-first must leave the committed object readable");
+        assert!(matches!(
+            set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+    }
+
+    async fn assert_abort_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x48; 4096], &create_opts).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone()));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let object_holder = set_disks
+            .new_ns_lock(bucket, object)
+            .await
+            .expect("object namespace lock should be created")
+            .get_write_lock(Duration::from_secs(5))
+            .await
+            .expect("test should hold the object lock");
+        let holder = set_disks
+            .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path)
+            .await
+            .expect("upload namespace lock should be created")
+            .get_write_lock(Duration::from_secs(5))
+            .await
+            .expect("test should hold the upload lock");
+        signaling.wait_for_attempts(1).await;
+
+        let abort_store = set_disks.clone();
+        let abort_upload_id = upload_id.clone();
+        let abort = tokio::spawn(async move {
+            abort_store
+                .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                .await
+        });
+        drop(holder);
+
+        abort
+            .await
+            .expect("abort task should not panic")
+            .expect("abort should win the upload finalization");
+        drop(object_holder);
+        let complete_err = complete
+            .await
+            .expect("completion task should not panic")
+            .expect_err("completion must observe the aborted upload");
+        assert!(matches!(complete_err, StorageError::InvalidUploadID(..)));
+        let object_err = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect_err("abort-first must not publish an object");
+        assert!(matches!(object_err, StorageError::ObjectNotFound(..)));
+        assert!(matches!(
+            set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+    }
+
     async fn assert_quorum_minus_one_retry_preserves_completable_part(
         disk_count: usize,
         parity: usize,
@@ -3039,6 +3225,81 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn abort_and_complete_linearize_for_plain_sse_and_legacy_layouts() {
+        assert_complete_first_linearizes("multipart-complete-first-plain", "object", ObjectOptions::default()).await;
+        assert_abort_first_linearizes("multipart-abort-first-plain", "object", ObjectOptions::default()).await;
+
+        let encrypted_opts = ObjectOptions {
+            user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]),
+            ..Default::default()
+        };
+        temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async {
+            assert_complete_first_linearizes("multipart-complete-first-sse", "object", encrypted_opts.clone()).await;
+            assert_abort_first_linearizes("multipart-abort-first-sse", "object", encrypted_opts.clone()).await;
+        })
+        .await;
+        temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("false"))], async {
+            assert_complete_first_linearizes("multipart-complete-first-legacy", "object", encrypted_opts.clone()).await;
+            assert_abort_first_linearizes("multipart-abort-first-legacy", "object", encrypted_opts).await;
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn abort_enforces_delete_write_quorum_boundary() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-abort-delete-quorum";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let quorum_upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("multipart upload should be created");
+
+        let saved_disks = {
+            let mut disks = set_disks.disks.write().await;
+            let saved = disks.clone();
+            disks[3] = None;
+            saved
+        };
+        set_disks
+            .abort_multipart_upload(bucket, object, &quorum_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("abort should succeed at the exact delete write quorum");
+        *set_disks.disks.write().await = saved_disks;
+        assert!(matches!(
+            set_disks
+                .check_upload_id_exists(bucket, object, &quorum_upload.upload_id, false)
+                .await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+
+        let below_quorum_upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("second multipart upload should be created");
+        let saved_disks = {
+            let mut disks = set_disks.disks.write().await;
+            let saved = disks.clone();
+            disks[2] = None;
+            disks[3] = None;
+            saved
+        };
+        let err = set_disks
+            .abort_multipart_upload(bucket, object, &below_quorum_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect_err("abort must report a delete below write quorum");
+        assert!(matches!(err, StorageError::ErasureWriteQuorum));
+
+        *set_disks.disks.write().await = saved_disks;
+        set_disks
+            .check_upload_id_exists(bucket, object, &below_quorum_upload.upload_id, false)
+            .await
+            .expect("failed abort must leave quorum-visible staging on the restored disks");
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_revalidates_layout_candidate_after_upload_lock() {
@@ -3170,6 +3431,17 @@ mod tests {
             tokio::task::yield_now().await;
             assert!(!abort.is_finished(), "abort must wait until completion releases the upload lock");
 
+            let list_store = set_disks.clone();
+            let list_upload_id = upload_id.clone();
+            let list = tokio::spawn(async move {
+                list_store
+                    .list_object_parts(bucket, object, &list_upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default())
+                    .await
+            });
+            signaling.wait_for_attempts(3).await;
+            tokio::task::yield_now().await;
+            assert!(!list.is_finished(), "ListParts must wait until completion releases the upload lock");
+
             barrier.release();
             complete
                 .await
@@ -3180,10 +3452,68 @@ mod tests {
                 .expect("abort task should not panic")
                 .expect_err("the committed upload should no longer exist when abort acquires the lock");
             assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
+            let list_err = list
+                .await
+                .expect("ListParts task should not panic")
+                .expect_err("the committed upload should no longer exist when ListParts acquires the lock");
+            assert!(matches!(list_err, StorageError::InvalidUploadID(..)));
         })
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn complete_validates_parts_after_an_inflight_upload_part_commit() {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        let bucket = "multipart-complete-put-part-race-bucket";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, original_parts) =
+            stage_upload_with_create_opts(&set_disks, bucket, object, &[0x49; 4096], &ObjectOptions::default()).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
+
+        let put_store = set_disks.clone();
+        let put_upload_id = upload_id.clone();
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![0x4a; 4096]);
+            put_store
+                .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
+                .await
+        });
+        barrier.wait_until_paused().await;
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, original_parts, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+        tokio::task::yield_now().await;
+        assert!(!complete.is_finished(), "completion must wait for the UploadPart commit lock");
+
+        barrier.release();
+        put.await
+            .expect("UploadPart task should not panic")
+            .expect("UploadPart replacement should commit");
+        let err = complete
+            .await
+            .expect("completion task should not panic")
+            .expect_err("completion must reject the stale ETag after UploadPart wins");
+        assert!(matches!(err, StorageError::InvalidPart(..)));
+        set_disks
+            .list_object_parts(bucket, object, &upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default())
+            .await
+            .expect("failed completion must leave the upload retryable");
+    }
+
     #[tokio::test(start_paused = true)]
     #[serial]
     async fn complete_fences_upload_lock_loss_before_commit() {

crates/ecstore/src/set_disk/ops/object.rs

@@ -25,7 +25,10 @@ use crate::set_disk::read::GetObjectDownstreamWriter;
 
 use crate::bucket::lifecycle::{
     tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry},
-    tier_sweeper::{Jentry, RemoteTierDeleteOutcome, delete_object_from_remote_tier_with_lease_idempotent},
+    tier_sweeper::{
+        Jentry, RemoteTierDeleteOutcome, delete_confirmed_transition_candidate_exact_with_lease_idempotent,
+        delete_object_from_remote_tier_with_lease_idempotent,
+    },
     transition_transaction::{
         TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
         TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
@@ -1677,7 +1680,11 @@ pub(crate) async fn cleanup_uncommitted_transition_upload(
     cleanup_version: &str,
     version_id_exact: bool,
 ) -> std::io::Result<RemoteTierDeleteOutcome> {
-    delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, version_id_exact).await
+    if version_id_exact {
+        delete_confirmed_transition_candidate_exact_with_lease_idempotent(object, cleanup_version, lease).await
+    } else {
+        delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, false).await
+    }
 }
 
 fn log_transition_upload_cleanup_failure(lease: &TierOperationLease, object: &str, cleanup_version: &str, err: &std::io::Error) {
@@ -1814,7 +1821,7 @@ impl Drop for TransitionUploadCleanup {
     }
 }
 
-async fn cleanup_rejected_transition_upload_durably(
+pub(crate) async fn cleanup_rejected_transition_upload_durably(
     lease: &TierOperationLease,
     object: &str,
     cleanup_version: &str,
@@ -1827,6 +1834,13 @@ async fn cleanup_rejected_transition_upload_durably(
         tier_name: lease.tier_name().to_string(),
         backend_identity: Some(lease.backend_identity()),
         version_id_exact,
+        version_state: if !version_id_exact {
+            rustfs_filemeta::TransitionVersionState::KnownDisabled
+        } else if cleanup_version == "null" {
+            rustfs_filemeta::TransitionVersionState::SuspendedNull
+        } else {
+            rustfs_filemeta::TransitionVersionState::Exact
+        },
     };
 
     let journal_error = if let Some(api) = api.as_ref() {
@@ -1986,12 +2000,83 @@ async fn advance_and_save_transition_transaction(
     next: TransitionTransactionState,
     remote_version: Option<TransitionRemoteVersion>,
 ) -> Result<()> {
+    #[cfg(test)]
+    record_transition_uploaded_save_attempt(transaction, next);
     transaction
         .advance(transaction.fence(), next, remote_version)
         .map_err(Error::other)?;
     save_transition_transaction_if_available(api, transaction).await
 }
 
+#[cfg(test)]
+struct TransitionUploadedSaveProbeState {
+    bucket: String,
+    object: String,
+    attempts: std::sync::atomic::AtomicUsize,
+}
+
+#[cfg(test)]
+struct TransitionUploadedSaveProbe {
+    state: Arc<TransitionUploadedSaveProbeState>,
+}
+
+#[cfg(test)]
+static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Option<Arc<TransitionUploadedSaveProbeState>>>> =
+    std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl TransitionUploadedSaveProbe {
+    fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(TransitionUploadedSaveProbeState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            attempts: std::sync::atomic::AtomicUsize::new(0),
+        });
+        let mut slot = TRANSITION_UPLOADED_SAVE_PROBE
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition uploaded-save probe mutex should not poison");
+        assert!(slot.is_none(), "transition uploaded-save probe must be installed by one test at a time");
+        *slot = Some(Arc::clone(&state));
+        drop(slot);
+        Self { state }
+    }
+
+    fn attempts(&self) -> usize {
+        self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
+    }
+}
+
+#[cfg(test)]
+impl Drop for TransitionUploadedSaveProbe {
+    fn drop(&mut self) {
+        let mut slot = TRANSITION_UPLOADED_SAVE_PROBE
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition uploaded-save probe mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction, next: TransitionTransactionState) {
+    if next != TransitionTransactionState::Uploaded {
+        return;
+    }
+    let state = TRANSITION_UPLOADED_SAVE_PROBE
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("transition uploaded-save probe mutex should not poison")
+        .as_ref()
+        .filter(|state| state.bucket == transaction.source.bucket && state.object == transaction.source.object)
+        .cloned();
+    if let Some(state) = state {
+        state.attempts.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+    }
+}
+
 async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
     if let Some(api) = api {
         return delete_transition_transaction_record(api.clone(), transaction_id).await;
@@ -2242,11 +2327,52 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
     }
 }
 
-fn parse_transition_version_id(remote_version: &str) -> std::result::Result<Option<Uuid>, uuid::Error> {
+fn persisted_transition_version(
+    remote_version: &str,
+) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
+    persisted_transition_version_with_gate(remote_version, remote_version_state_writer_enabled())
+}
+
+fn remote_version_state_writer_enabled() -> bool {
+    remote_version_state_writer_enabled_for(
+        rustfs_utils::get_env_bool(
+            rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE,
+            rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE,
+        ),
+        rustfs_utils::get_env_bool(
+            rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+            rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+        ),
+    )
+}
+
+fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool) -> bool {
+    requested && fleet_confirmed
+}
+
+fn persisted_transition_version_with_gate(
+    remote_version: &str,
+    remote_version_state_writer_enabled: bool,
+) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
     if remote_version.is_empty() {
-        return Ok(None);
+        return Ok((None, rustfs_filemeta::TransitionVersionState::KnownDisabled));
+    }
+
+    match Uuid::parse_str(remote_version) {
+        Ok(version_id) if version_id.is_nil() => Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier returned a nil object version ID",
+        )),
+        Ok(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
+        Err(_) if !remote_version_state_writer_enabled => Err(std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "opaque remote tier versions require the operator-attested fleet gate",
+        )),
+        Err(_) if remote_version == "null" => {
+            Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::SuspendedNull))
+        }
+        Err(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
     }
-    Uuid::parse_str(remote_version).map(|version_id| (!version_id.is_nil()).then_some(version_id))
 }
 
 #[cfg(test)]
@@ -2484,16 +2610,20 @@ mod transition_upload_completion_tests {
 
 #[cfg(test)]
 mod transition_version_id_tests {
-    use super::{TransitionUploadCandidate, parse_transition_version_id};
+    use super::{
+        TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate,
+        remote_version_state_writer_enabled_for,
+    };
+    use rustfs_filemeta::TransitionVersionState;
     use uuid::Uuid;
 
     #[test]
     fn normalizes_persisted_unversioned_ids_and_preserves_put_constraints() {
-        assert_eq!(parse_transition_version_id("").expect("empty remote version should be valid"), None);
         assert_eq!(
-            parse_transition_version_id(&Uuid::nil().to_string()).expect("nil remote version should be valid"),
-            None
+            persisted_transition_version("").expect("empty remote version identifies an unversioned tier"),
+            (None, TransitionVersionState::KnownDisabled)
         );
+        assert!(persisted_transition_version(&Uuid::nil().to_string()).is_err());
         let nil_put_response = Uuid::nil().to_string();
         let nil_candidate = TransitionUploadCandidate::from_put_response(nil_put_response.clone());
         assert_eq!(nil_candidate.cleanup_version(), nil_put_response);
@@ -2505,12 +2635,14 @@ mod transition_version_id_tests {
     }
 
     #[test]
-    fn preserves_valid_remote_id_and_rejects_invalid_text() {
+    fn preserves_uuid_and_gates_opaque_remote_ids() {
         let version_id = Uuid::new_v4();
         assert_eq!(
-            parse_transition_version_id(&version_id.to_string()).expect("UUID remote version should be valid"),
-            Some(version_id)
+            persisted_transition_version(&version_id.to_string()).expect("UUID remote version"),
+            (Some(version_id.to_string()), TransitionVersionState::Exact)
         );
+        assert!(persisted_transition_version("null").is_err());
+        assert!(persisted_transition_version("opaque-version-token").is_err());
         assert_eq!(
             TransitionUploadCandidate::from_put_response(version_id.to_string()).cleanup_version(),
             version_id.to_string()
@@ -2519,7 +2651,44 @@ mod transition_version_id_tests {
             TransitionUploadCandidate::from_put_response("opaque-version-token".to_string()).cleanup_version(),
             "opaque-version-token"
         );
-        assert!(parse_transition_version_id("not-a-uuid").is_err());
+    }
+
+    #[test]
+    fn remote_version_state_writer_requires_request_and_fleet_confirmation() {
+        for (case, requested, fleet_confirmed, expected) in [
+            ("old defaults", false, false, false),
+            ("missing fleet confirmation", true, false, false),
+            ("missing local opt-in", false, true, false),
+            ("explicitly unconfirmed fleet", true, false, false),
+            ("rolled-back writer", false, true, false),
+            ("fully upgraded fleet", true, true, true),
+        ] {
+            assert_eq!(remote_version_state_writer_enabled_for(requested, fleet_confirmed), expected, "{case}");
+        }
+    }
+
+    #[test]
+    fn fleet_gate_enables_null_and_opaque_remote_version_states() {
+        for (remote_version, expected) in [
+            ("null", (Some("null".to_string()), TransitionVersionState::SuspendedNull)),
+            (
+                "opaque-version-token",
+                (Some("opaque-version-token".to_string()), TransitionVersionState::Exact),
+            ),
+        ] {
+            assert!(
+                persisted_transition_version_with_gate(remote_version, false).is_err(),
+                "missing fleet confirmation must reject {remote_version:?}"
+            );
+            assert_eq!(
+                persisted_transition_version_with_gate(remote_version, true).expect("fleet-confirmed state must be persisted"),
+                expected
+            );
+        }
+        assert_eq!(
+            persisted_transition_version_with_gate("", true).expect("empty remote version identifies an unversioned tier"),
+            (None, TransitionVersionState::KnownDisabled)
+        );
     }
 }
 
@@ -3789,6 +3958,20 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
             return Err(err.into());
         }
+        let (transition_version_id, transition_version_state) = match persisted_transition_version(candidate.remote_version()) {
+            Ok(version) => version,
+            Err(err) => {
+                let cleanup_api = transition_cleanup_store(&self.ctx).await;
+                if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await {
+                    return Err(StorageError::Io(std::io::Error::other(format!(
+                        "{err}; rejected remote upload cleanup failed: {cleanup_err}"
+                    ))));
+                }
+                delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
+                    .await;
+                return Err(err.into());
+            }
+        };
         if let Err(err) = advance_and_save_transition_transaction(
             transaction_api.as_ref(),
             &mut transaction,
@@ -3806,16 +3989,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
             return Err(err);
         }
-        let transition_version_id = match parse_transition_version_id(candidate.remote_version()) {
-            Ok(version_id) => version_id,
-            Err(err) => {
-                if upload_cleanup.cleanup().await.is_ok() {
-                    delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
-                        .await;
-                }
-                return Err(err.into());
-            }
-        };
 
         let mut commit_opts = opts.clone();
         commit_opts.no_lock = true;
@@ -3873,7 +4046,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         current_fi.transition_status = TRANSITION_COMPLETE.to_string();
         current_fi.transitioned_objname = dest_obj;
         current_fi.transition_tier = opts.transition.tier.clone();
-        current_fi.transition_version_id = transition_version_id;
+        current_fi.transition_version_id = transition_version_id
+            .as_deref()
+            .and_then(|version_id| Uuid::parse_str(version_id).ok());
+        current_fi.transition_version = transition_version_id;
+        current_fi.transition_version_state = transition_version_state;
         rustfs_utils::http::metadata_compat::insert_str(
             &mut current_fi.metadata,
             rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
@@ -4090,6 +4267,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             let mut p_reader = PutObjReader::new(hash_reader);
             return match self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await {
                 Ok(restored_info) => {
+                    let restored_info = self_.finalize_restore_metadata(bucket, object, &restored_info, &opts).await?;
                     send_event(EventArgs {
                         event_name: EventName::ObjectRestoreCompleted.as_str().to_string(),
                         bucket_name: bucket.to_string(),
@@ -4224,6 +4402,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 return set_restore_header_fn(&mut oi, Some(err)).await;
             }
         };
+        let restored_info = self_.finalize_restore_metadata(bucket, object, &restored_info, opts).await?;
         send_event(EventArgs {
             event_name: EventName::ObjectRestoreCompleted.as_str().to_string(),
             bucket_name: bucket.to_string(),
@@ -4737,6 +4916,33 @@ mod transition_commit_failure_tests {
     }
 
     #[tokio::test]
+    async fn rejected_unsupported_remote_versions_are_cleaned_up() {
+        for remote_version in ["null", "opaque-version-token"] {
+            let manager = TierConfigMgr::new();
+            let backend = register_mock_tier(&manager, "WARM").await;
+            let lease = TierConfigMgr::acquire_operation_lease(&manager, "WARM")
+                .await
+                .expect("mock tier lease should be available");
+            let candidate = TransitionUploadCandidate::from_put_response(remote_version.to_string());
+
+            persisted_transition_version(candidate.remote_version()).expect_err("unsupported writer version must fail closed");
+            cleanup_rejected_transition_upload_durably(
+                &lease,
+                "remote/object",
+                candidate.cleanup_version(),
+                candidate.cleanup_version_is_exact(),
+                None,
+            )
+            .await
+            .expect("rejected remote upload must be cleaned up");
+
+            assert_eq!(
+                backend.remove_versions().await,
+                vec![("remote/object".to_string(), candidate.cleanup_version().to_string())]
+            );
+        }
+    }
+
     #[serial_test::serial(restore_multipart_failure_point)]
     async fn multipart_restore_aborts_every_post_create_failure() {
         let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -6684,6 +6890,45 @@ mod transition_upload_integrity_tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn unversioned_remote_version_is_persisted_without_version_id() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-unversioned-tier-bucket";
+        let object = "object.bin";
+        let payload = b"unversioned remote tier must commit without a version id".repeat(1024);
+        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        backend.set_put_remote_version(Some(String::new())).await;
+        let save_probe = TransitionUploadedSaveProbe::install(bucket, object);
+
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("an unversioned remote version must commit");
+        let (fi, _, _) = set_disks
+            .get_object_fileinfo(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    metadata_cache_safe: false,
+                    ..Default::default()
+                },
+                true,
+                false,
+            )
+            .await
+            .expect("committed unversioned transition metadata should be readable");
+        assert_eq!(fi.transition_version_id, None);
+        assert_eq!(fi.transition_version, None);
+        assert_eq!(fi.transition_version_state, rustfs_filemeta::TransitionVersionState::KnownDisabled);
+        assert_eq!(save_probe.attempts(), 1);
+        assert_eq!(backend.remove_count().await, 0);
+        assert_eq!(backend.object_count().await, 1);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn opaque_remote_version_is_cleaned_before_parse_failure() {
@@ -6699,7 +6944,7 @@ mod transition_upload_integrity_tests {
         set_disks
             .transition_object(bucket, object, &transition_options(&original, tier_name))
             .await
-            .expect_err("an unparseable remote version must fail closed");
+            .expect_err("an opaque remote version must fail closed until the capability gate is active");
         let removed_versions = backend.remove_versions().await;
         assert_eq!(removed_versions.len(), 1);
         assert_eq!(removed_versions[0].1, "opaque-version-token");
@@ -6707,6 +6952,38 @@ mod transition_upload_integrity_tests {
         assert_local_source_intact(&set_disks, bucket, object, &payload).await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn nil_remote_version_is_cleaned_exactly_before_transaction_persistence() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-nil-version-bucket";
+        let object = "object.bin";
+        let payload = b"nil remote version must retain local data".repeat(1024);
+        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let remote_version = Uuid::nil().to_string();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        backend.set_put_remote_version(Some(remote_version.clone())).await;
+        let save_probe = TransitionUploadedSaveProbe::install(bucket, object);
+
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect_err("a nil remote version must fail closed before transaction persistence");
+        let put_versions = backend.put_versions().await;
+        let removed_versions = backend.remove_versions().await;
+        assert_eq!(removed_versions, put_versions);
+        assert_eq!(removed_versions.len(), 1);
+        assert_eq!(
+            removed_versions.first().map(|(_, version)| version.as_str()),
+            Some(remote_version.as_str())
+        );
+        assert_eq!(save_probe.attempts(), 0, "nil remote version must be rejected before saving Uploaded");
+        assert_eq!(backend.exact_remove_count(), 1);
+        assert_eq!(backend.object_count().await, 0);
+        assert_local_source_intact(&set_disks, bucket, object, &payload).await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn authoritative_read_failure_after_upload_cleans_exact_candidate_and_preserves_source() {
@@ -7054,23 +7331,36 @@ mod transition_source_identity_matrix_tests {
             let object = format!("identity-{index}.bin");
             let payload = vec![u8::try_from(index + 1).expect("matrix index should fit u8"); 1024 * 1024];
             let mut reader = PutObjReader::from_vec(payload);
+            let source_version_id = Uuid::new_v4();
+            let source_opts = ObjectOptions {
+                version_id: Some(source_version_id.to_string()),
+                versioned: true,
+                ..Default::default()
+            };
             let original = set_disks
-                .put_object(bucket, &object, &mut reader, &ObjectOptions::default())
+                .put_object(bucket, &object, &mut reader, &source_opts)
                 .await
                 .expect("source object should be written");
             let (source, _, _) = set_disks
-                .get_object_fileinfo(bucket, &object, &ObjectOptions::default(), true, false)
+                .get_object_fileinfo(bucket, &object, &source_opts, true, false)
                 .await
                 .expect("source metadata should resolve");
+            assert_eq!(source.version_id, Some(source_version_id));
+            assert_eq!(
+                transition_source_identity(bucket, &object, &source, &source_opts, &get_raw_etag(&source.metadata))
+                    .expect("persisted versioned source identity should build")
+                    .version_mode,
+                TransitionSourceVersionMode::Versioned
+            );
             let opts = ObjectOptions {
                 no_lock: true,
+                versioned: true,
                 transition: TransitionOptions {
                     status: TRANSITION_PENDING.to_string(),
                     tier: tier_name.clone(),
                     etag: original.etag.clone().unwrap_or_default(),
                     ..Default::default()
                 },
-                version_id: original.version_id.map(|version| version.to_string()),
                 mod_time: original.mod_time,
                 ..Default::default()
             };
@@ -7083,7 +7373,10 @@ mod transition_source_identity_matrix_tests {
 
             let mut changed = source.clone();
             match field {
-                IdentityField::VersionId => changed.version_id = Some(Uuid::new_v4()),
+                IdentityField::VersionId => {
+                    changed.version_id = Some(Uuid::new_v4());
+                    changed.fresh = true;
+                }
                 IdentityField::DataDir => changed.data_dir = Some(Uuid::new_v4()),
                 IdentityField::ModTime => {
                     changed.mod_time = changed.mod_time.map(|value| value + time::Duration::nanoseconds(1));
@@ -7093,17 +7386,27 @@ mod transition_source_identity_matrix_tests {
                     changed.metadata.insert("etag".to_string(), format!("changed-{index}"));
                 }
             }
+            // Replace the object version list so VersionId drift removes the
+            // accepted source version instead of appending a second version.
+            changed.fresh = true;
             for disk in &disk_stores {
                 disk.write_metadata("", bucket, &object, changed.clone())
                     .await
                     .expect("single-field metadata drift should be written");
             }
+            let persisted_opts = ObjectOptions {
+                version_id: changed.version_id.map(|version_id| version_id.to_string()),
+                versioned: true,
+                ..Default::default()
+            };
+            let (persisted, _, _) = set_disks
+                .get_object_fileinfo(bucket, &object, &persisted_opts, true, false)
+                .await
+                .expect("drifted source metadata should resolve");
             put_barrier.release();
 
-            transition
-                .await
-                .expect("transition task should not panic")
-                .expect_err("transition must reject a source whose identity changed after upload");
+            let result = transition.await.expect("transition task should not panic");
+            assert!(result.is_err(), "transition must reject {field:?} drift");
             let expected_attempts = index + 1;
             assert_eq!(backend.put_count().await, expected_attempts);
             assert_eq!(backend.remove_count().await, expected_attempts);
@@ -7114,26 +7417,26 @@ mod transition_source_identity_matrix_tests {
             );
 
             match field {
-                IdentityField::VersionId => assert_ne!(source.version_id, changed.version_id),
-                IdentityField::DataDir => assert_ne!(source.data_dir, changed.data_dir),
-                IdentityField::ModTime => assert_ne!(source.mod_time, changed.mod_time),
-                IdentityField::Size => assert_ne!(source.size, changed.size),
-                IdentityField::Etag => assert_ne!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata)),
+                IdentityField::VersionId => assert_ne!(source.version_id, persisted.version_id),
+                IdentityField::DataDir => assert_ne!(source.data_dir, persisted.data_dir),
+                IdentityField::ModTime => assert_ne!(source.mod_time, persisted.mod_time),
+                IdentityField::Size => assert_ne!(source.size, persisted.size),
+                IdentityField::Etag => assert_ne!(get_raw_etag(&source.metadata), get_raw_etag(&persisted.metadata)),
             }
             if !matches!(field, IdentityField::VersionId) {
-                assert_eq!(source.version_id, changed.version_id);
+                assert_eq!(source.version_id, persisted.version_id);
             }
             if !matches!(field, IdentityField::DataDir) {
-                assert_eq!(source.data_dir, changed.data_dir);
+                assert_eq!(source.data_dir, persisted.data_dir);
             }
             if !matches!(field, IdentityField::ModTime) {
-                assert_eq!(source.mod_time, changed.mod_time);
+                assert_eq!(source.mod_time, persisted.mod_time);
             }
             if !matches!(field, IdentityField::Size) {
-                assert_eq!(source.size, changed.size);
+                assert_eq!(source.size, persisted.size);
             }
             if !matches!(field, IdentityField::Etag) {
-                assert_eq!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata));
+                assert_eq!(get_raw_etag(&source.metadata), get_raw_etag(&persisted.metadata));
             }
         }
     }

crates/ecstore/src/set_disk/read.rs

@@ -2016,11 +2016,15 @@ mod metadata_cache_tests {
         fi.size = 1;
         fi.erasure.index = 1;
         fi.metadata.insert("etag".to_string(), "etag-1".to_string());
+        fi.add_object_part(1, "part-etag".to_string(), 1, fi.mod_time, 1, None, None);
         fi
     }
 
     #[tokio::test]
     async fn get_object_with_fileinfo_rejects_positive_size_without_parts() {
+        let mut fi = valid_test_fileinfo("object");
+        fi.parts.clear();
+
         let mut output = Vec::new();
         let err = SetDisks::get_object_with_fileinfo(
             "bucket",
@@ -2028,7 +2032,7 @@ mod metadata_cache_tests {
             0,
             1,
             &mut output,
-            valid_test_fileinfo("object"),
+            fi,
             Vec::new(),
             &[],
             0,
@@ -2119,12 +2123,6 @@ mod metadata_cache_tests {
 
         let mut invalid_erasure = valid_test_fileinfo(object);
         invalid_erasure.erasure.block_size = 0;
-        invalid_erasure.parts.push(ObjectPartInfo {
-            number: 1,
-            size: 1,
-            actual_size: 1,
-            ..Default::default()
-        });
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
@@ -2159,6 +2157,7 @@ mod metadata_cache_tests {
         let object = "empty";
         let mut fi = valid_test_fileinfo(object);
         fi.size = 0;
+        fi.parts.clear();
 
         let mut output = Vec::new();
         SetDisks::get_object_with_fileinfo(
@@ -2191,12 +2190,6 @@ mod metadata_cache_tests {
         let mut fi = valid_test_fileinfo(object);
         fi.erasure.block_size = 1;
         fi.erasure.distribution = vec![1, 2, 3, 4];
-        fi.parts.push(ObjectPartInfo {
-            number: 1,
-            size: 1,
-            actual_size: 1,
-            ..Default::default()
-        });
 
         let mut output = Vec::new();
         let err = SetDisks::get_object_with_fileinfo(

crates/ecstore/src/set_disk/replication.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use super::*;
+use crate::bucket::lifecycle::lifecycle;
+use rustfs_filemeta::RestoreStatusOps;
 use rustfs_utils::http::headers::{AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE};
+use s3s::dto::{RestoreStatus, Timestamp};
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 struct RestoreCleanupIdentity {
@@ -43,6 +46,69 @@ impl RestoreCleanupIdentity {
 }
 
 impl SetDisks {
+    pub(super) async fn finalize_restore_metadata(
+        &self,
+        bucket: &str,
+        object: &str,
+        obj_info: &ObjectInfo,
+        opts: &ObjectOptions,
+    ) -> Result<ObjectInfo> {
+        let expected = RestoreCleanupIdentity::from_object_info(obj_info);
+        let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?;
+        let expected_etag = obj_info
+            .etag
+            .clone()
+            .unwrap_or_else(|| get_raw_etag(obj_info.user_defined.as_ref()));
+        let version_id = expected.version_id.map(|v| v.to_string());
+        let _lock_guard = if !opts.no_lock {
+            Some(
+                self.acquire_write_lock_diag("restore_finalize_metadata", bucket, object)
+                    .await?,
+            )
+        } else {
+            None
+        };
+        let read_opts = ObjectOptions {
+            version_id,
+            versioned: opts.versioned,
+            version_suspended: opts.version_suspended,
+            ..Default::default()
+        };
+        let (mut fi, _, disks) = self
+            .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
+            .await?;
+        if let Some(expected_operation_id) = expected_operation_id {
+            require_restore_operation_id(&fi.metadata, expected_operation_id)?;
+        }
+        if !expected.matches_file_info(&fi, &expected_etag) {
+            return Err(Error::other("restored object changed before restore metadata finalization"));
+        }
+        let restore_expiry =
+            lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1));
+        fi.metadata.insert(
+            X_AMZ_RESTORE.as_str().to_string(),
+            RestoreStatus {
+                is_restore_in_progress: Some(false),
+                restore_expiry_date: Some(Timestamp::from(restore_expiry)),
+            }
+            .to_string(),
+        );
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        self.update_object_meta_with_opts(
+            bucket,
+            object,
+            fi.clone(),
+            disks.as_slice(),
+            &UpdateMetadataOpts {
+                replace_user_metadata: true,
+                ..Default::default()
+            },
+        )
+        .await?;
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
+    }
+
     pub async fn update_restore_metadata(
         &self,
         bucket: &str,

crates/ecstore/src/set_disk/transition_matrix_tests.rs

@@ -13,10 +13,11 @@
 // limitations under the License.
 
 use super::*;
-use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions};
+use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time};
 use crate::ecstore_validation_blackbox::make_local_set_disks;
 use crate::services::tier::test_util::register_mock_tier;
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+use rustfs_filemeta::{RestoreStatusOps as _, parse_restore_obj_status};
 use tokio::io::AsyncReadExt;
 
 async fn prime_metadata_generation(set_disks: &SetDisks, bucket: &str, object: &str) -> GetObjectMetadataCacheKey {
@@ -83,13 +84,57 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() {
     let transitioned_generation = prime_metadata_generation(&set_disks, bucket, object).await;
     let mut restore_opts = ObjectOptions::default();
     restore_opts.transition.restore_request.days = Some(1);
-    Arc::clone(&set_disks)
-        .restore_transitioned_object(bucket, object, &restore_opts)
-        .await
-        .expect("restore should succeed");
+    let restore_started = OffsetDateTime::now_utc();
+    let expiry_from_restore_start = temp_env::async_with_vars(
+        [
+            ("RUSTFS_ILM_DEBUG_DAY_SECS", Some("1")),
+            ("RUSTFS_ILM_PROCESS_TIME", Some("1")),
+        ],
+        async {
+            let expiry_from_restore_start = expected_expiry_time(restore_started, 1);
+            let get_barrier = backend.arm_get_barrier().await;
+            let restore_set = Arc::clone(&set_disks);
+            let restore =
+                tokio::spawn(async move { restore_set.restore_transitioned_object(bucket, object, &restore_opts).await });
+            get_barrier.wait_until_paused().await;
+            tokio::time::timeout(Duration::from_secs(5), async {
+                loop {
+                    if expected_expiry_time(OffsetDateTime::now_utc(), 1) > expiry_from_restore_start {
+                        break;
+                    }
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .expect("test clock should cross the next accelerated lifecycle boundary");
+            get_barrier.release();
+            restore
+                .await
+                .expect("restore task should join")
+                .expect("restore should succeed");
+            expiry_from_restore_start
+        },
+    )
+    .await;
     assert_generation_reclaimed(&set_disks, &transitioned_generation).await;
     assert_eq!(backend.get_count().await, 1, "restore should read the remote candidate exactly once");
 
+    let restored_info = set_disks
+        .get_object_info(bucket, object, &ObjectOptions::default())
+        .await
+        .expect("restored object metadata should be readable");
+    let restore_status = parse_restore_obj_status(
+        restored_info
+            .user_defined
+            .get(s3s::header::X_AMZ_RESTORE.as_str())
+            .expect("completed restore header should be present"),
+    )
+    .expect("completed restore header should parse");
+    assert!(
+        restore_status.expiry().expect("completed restore should have an expiry") > expiry_from_restore_start,
+        "restore expiry must be based on completion, not the time the remote copy started"
+    );
+
     let mut restored = Vec::new();
     set_disks
         .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())