fix(ilm): recover orphaned restore generations (#7104)

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -4887,6 +4887,7 @@ pub async fn put_restore_opts(
         meta.insert(k.to_string(), v.clone());
     }
     rustfs_utils::http::metadata_compat::remove_str(&mut meta, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID);
+    rustfs_utils::http::metadata_compat::remove_str(&mut meta, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_WORKER_LOCK);
     if !oi.user_tags.is_empty() {
         meta.insert(AMZ_OBJECT_TAGGING.to_string(), (*oi.user_tags).clone());
     }

crates/ecstore/src/services/tier/tier.rs

@@ -3168,6 +3168,10 @@ impl TierConfigMgr {
         lease
     }
 
+    pub(crate) fn operation_lease_blocked_by_mutation(err: &AdminError) -> bool {
+        err.message == TIER_MUTATION_BLOCK_MESSAGE
+    }
+
     async fn admin_update_lock(handle: &Arc<RwLock<Self>>) -> tokio::sync::OwnedMutexGuard<()> {
         let update_lock = {
             let manager = handle.read().await;
@@ -5896,6 +5900,24 @@ mod tests {
         TransitionStorageClass,
     };
 
+    #[test]
+    fn restore_retry_classifier_matches_only_the_exact_mutation_block() {
+        assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&AdminError::msg(
+            TIER_MUTATION_BLOCK_MESSAGE,
+        )));
+        for message in [
+            "Remote tier configuration is being replaced: COLD",
+            "remote tier configuration is being replaced",
+            "Remote tier configuration is unavailable",
+            "",
+        ] {
+            assert!(
+                !TierConfigMgr::operation_lease_blocked_by_mutation(&AdminError::msg(message)),
+                "unrelated error must not consume the restore retry budget: {message}"
+            );
+        }
+    }
+
     struct SetupTypeGuard {
         previous: SetupType,
     }

crates/ecstore/src/set_disk/mod.rs

@@ -153,8 +153,8 @@ use rustfs_utils::http::headers::{
 };
 use rustfs_utils::http::{
     SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE,
-    SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, contains_key_str, get_header_map, get_str, insert_str,
-    is_object_encryption_marker, remove_header_map,
+    SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, contains_key_str, get_header_map,
+    get_str, insert_str, is_object_encryption_marker, remove_header_map,
 };
 use rustfs_utils::{
     HashAlgorithm,

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

@@ -35,14 +35,15 @@ use super::super::{
     OBJECT_OP_IGNORED_ERRS, ObjectInfo, ObjectLockDiagGuard, ObjectOptions, ObjectPartInfo, OffsetDateTime, PartInfo,
     PutObjReader, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY,
     Result, SLASH_SEPARATOR, SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID,
-    SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError,
-    Uuid, WriteLayout, check_object_lock_for_deletion_with_state, classify_multipart_part_write_path, coding,
-    complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, completed_multipart_object_part,
-    contains_key_str, create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str,
-    is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size, log_multipart_write_quorum_failure,
-    parts_after_marker, path_join_buf, record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs,
-    remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size,
-    strip_internal_multipart_metadata, to_object_err, warn,
+    SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SetDisks,
+    SmallWritePath, StorageError, Uuid, WriteLayout, check_object_lock_for_deletion_with_state,
+    classify_multipart_part_write_path, coding, complete_multipart_part_error, complete_multipart_part_error_result,
+    complete_part_checksum, completed_multipart_object_part, contains_key_str, create_bitrot_writer, debug, disk, error,
+    get_complete_multipart_md5, get_header_map, get_str, insert_str, is_err_object_not_found, is_err_version_not_found,
+    is_min_allowed_part_size, log_multipart_write_quorum_failure, parts_after_marker, path_join_buf,
+    record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs, remove_header_map, resolve_write_layout,
+    restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, strip_internal_multipart_metadata,
+    to_object_err, warn,
 };
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
 #[cfg(test)]
@@ -2437,6 +2438,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX);
         if expected_restore_operation_id.is_some() {
             rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_RESTORE_OPERATION_ID);
+            rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_RESTORE_WORKER_LOCK);
         }
         if opts.versioned {
             fi.version_id = Some(

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

@@ -36,10 +36,10 @@ use super::super::{
     ObjectLockConfigSnapshot, ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin,
     PutObjReader, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReadPathPlan, ReaderImpl, ReplicateDecision,
     ReplicationObjectBridge, Result, SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE,
-    SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError,
-    TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE,
-    X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE, adaptive_duplex_buffer_size, build_get_object_info,
-    build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
+    SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SetDisks,
+    SmallWritePath, StorageError, TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD,
+    X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE, adaptive_duplex_buffer_size,
+    build_get_object_info, build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
     check_object_lock_delete, check_object_lock_for_deletion_with_state, check_object_lock_retention_update,
     classify_get_codec_streaming_object_class, classify_put_write_path, classify_storage_error,
     collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug, delete_file_info_version_id,
@@ -96,6 +96,7 @@ const GET_MID_SIZE_STREAMING_MIN_SIZE: usize = 128 * 1024 + 1;
 const GET_MID_SIZE_STREAMING_MAX_SIZE: usize = 512 * 1024;
 const EVENT_LIFECYCLE_TRANSITION_CLEANUP: &str = "lifecycle_transition_cleanup";
 const EVENT_LIFECYCLE_TRANSITIONED_DELETE_CLEANUP_OWNER: &str = "lifecycle_transitioned_delete_cleanup_owner";
+const EVENT_LIFECYCLE_RESTORE_CLEANUP: &str = "lifecycle_restore_cleanup";
 
 fn replication_status_writeback_is_current(object_info: &ObjectInfo, condition: &ReplicationStatusWritebackCondition) -> bool {
     let current = object_info.replication_generation_snapshot();
@@ -321,7 +322,6 @@ use rustfs_utils::path::decode_dir_object;
 use rustfs_utils::path::encode_dir_object;
 use std::future::Future;
 use std::task::{Context, Poll};
-#[cfg(any(test, feature = "test-util"))]
 use std::time::Duration;
 #[cfg(test)]
 use tokio::io::AsyncReadExt;
@@ -1834,6 +1834,8 @@ fn is_restore_control_metadata(key: &str) -> bool {
         || key.eq_ignore_ascii_case(rustfs_utils::http::headers::AMZ_RESTORE_REQUEST_DATE)
         || rustfs_utils::http::internal_key_strip_suffix_prefix(key, SUFFIX_RESTORE_OPERATION_ID)
             .is_some_and(|remainder| remainder.is_empty())
+        || rustfs_utils::http::internal_key_strip_suffix_prefix(key, SUFFIX_RESTORE_WORKER_LOCK)
+            .is_some_and(|remainder| remainder.is_empty())
 }
 
 fn restore_metadata_update_preserves_protected_metadata(
@@ -1871,6 +1873,11 @@ mod restore_metadata_update_tests {
             SUFFIX_RESTORE_OPERATION_ID,
             Uuid::new_v4().to_string(),
         );
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut replacement,
+            SUFFIX_RESTORE_WORKER_LOCK,
+            rustfs_utils::http::metadata_compat::RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string(),
+        );
         assert!(restore_metadata_update_preserves_protected_metadata(&existing, &replacement));
 
         replacement.insert("x-amz-object-lock-mode".to_string(), "GOVERNANCE".to_string());
@@ -2107,6 +2114,42 @@ fn full_object_plaintext_len(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions
 }
 
 const RESTORE_MULTIPART_ABORT_FAILURES_TOTAL: &str = "rustfs_restore_multipart_abort_failures_total";
+const RESTORE_TIER_MUTATION_RETRY_BUDGET: Duration = Duration::from_secs(30);
+const RESTORE_TIER_MUTATION_RETRY_BASE: Duration = Duration::from_millis(250);
+const RESTORE_TIER_MUTATION_RETRY_CAP: Duration = Duration::from_secs(5);
+
+fn restore_tier_mutation_retry_delay(attempt: u32) -> Duration {
+    RESTORE_TIER_MUTATION_RETRY_BASE
+        .saturating_mul(1_u32 << attempt.min(5))
+        .min(RESTORE_TIER_MUTATION_RETRY_CAP)
+}
+
+async fn acquire_restore_tier_lease(
+    manager: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
+    tier_name: &str,
+    expected_backend_identity: Option<TierDestinationId>,
+) -> Result<TierOperationLease> {
+    let deadline = Instant::now() + RESTORE_TIER_MUTATION_RETRY_BUDGET;
+    let mut attempt = 0_u32;
+    loop {
+        let result = match expected_backend_identity {
+            Some(identity) => TierConfigMgr::acquire_operation_lease_for_backend_identity(manager, tier_name, identity).await,
+            None => TierConfigMgr::acquire_operation_lease(manager, tier_name).await,
+        };
+        match result {
+            Ok(lease) => return Ok(lease),
+            Err(err) if TierConfigMgr::operation_lease_blocked_by_mutation(&err) => {
+                let remaining = deadline.saturating_duration_since(Instant::now());
+                if remaining.is_zero() {
+                    return Err(Error::other(err));
+                }
+                tokio::time::sleep(restore_tier_mutation_retry_delay(attempt).min(remaining)).await;
+                attempt = attempt.saturating_add(1);
+            }
+            Err(err) => return Err(Error::other(err)),
+        }
+    }
+}
 
 #[cfg(test)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -3430,6 +3473,7 @@ impl SetDisks {
         }
         if expected_restore_operation_id.is_some() {
             rustfs_utils::http::metadata_compat::remove_str(&mut user_defined, SUFFIX_RESTORE_OPERATION_ID);
+            rustfs_utils::http::metadata_compat::remove_str(&mut user_defined, SUFFIX_RESTORE_WORKER_LOCK);
         }
         let WriteLayout {
             data_drives,
@@ -9295,15 +9339,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         //     _lock_guard = guard_opt;
         // }
         let self_ = self.clone();
-        let restore_header_self = self_.clone();
-        let set_restore_header_fn = async move |oi: &mut ObjectInfo, rerr: Option<Error>| -> Result<()> {
-            if rerr.is_none() {
-                return Ok(());
-            }
-            restore_header_self.update_restore_metadata(bucket, object, oi, opts).await?;
-            Err(rerr.unwrap())
-        };
-        let mut oi = ObjectInfo::default();
         let bucket_lifecycle_guard = if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id
             && opts.bucket_lifecycle_lock_fence.is_none()
         {
@@ -9331,247 +9366,251 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         }
         let mut restore_read_opts = opts.clone();
         restore_read_opts.include_part_checksums = true;
-        let fi = self
+        let actual = self
             .clone()
             .get_object_fileinfo(bucket, object, &restore_read_opts, true, false)
-            .await;
+            .await
+            .map_err(|err| to_object_err(err, vec![bucket, object]))?;
         drop(bucket_lifecycle_guard);
-        if let Err(err) = fi {
-            return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
-        }
-        let actual = fi?;
         let actual_fi = actual.fi();
 
-        oi = ObjectInfo::from_file_info(actual_fi, bucket, object, opts.versioned || opts.version_suspended);
-        // The tier reader releases its own lease at EOF, before PUT or
-        // CompleteMultipartUpload necessarily reaches metadata quorum. Keep a
-        // second exact-generation lease for the whole restore so the final
-        // same-key write lock and this lease together fence remote-tuple
-        // publication through commit.
-        let expected_backend_identity = tier_destination_id_from_metadata(&oi.user_defined).map_err(Error::other)?;
-        let tier_config_mgr = self.ctx.tier_config_mgr();
-        let _remote_tuple_publication_lease = match expected_backend_identity {
-            Some(identity) => {
-                TierConfigMgr::acquire_operation_lease_for_backend_identity(
-                    &tier_config_mgr,
-                    &oi.transitioned_object.tier,
-                    identity,
-                )
-                .await
+        let oi = ObjectInfo::from_file_info(actual_fi, bucket, object, opts.versioned || opts.version_suspended);
+        let restore_result: Result<()> = async {
+            // The tier reader releases its own lease at EOF, before PUT or
+            // CompleteMultipartUpload necessarily reaches metadata quorum. Keep a
+            // second exact-generation lease for the whole restore so the final
+            // same-key write lock and this lease together fence remote-tuple
+            // publication through commit.
+            let expected_backend_identity = tier_destination_id_from_metadata(&oi.user_defined).map_err(Error::other)?;
+            let tier_config_mgr = self.ctx.tier_config_mgr();
+            let _remote_tuple_publication_lease =
+                acquire_restore_tier_lease(&tier_config_mgr, &oi.transitioned_object.tier, expected_backend_identity).await?;
+            let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?;
+            if let Some(expected_operation_id) = expected_operation_id {
+                require_restore_operation_id(oi.user_defined.as_ref(), expected_operation_id)?;
             }
-            None => TierConfigMgr::acquire_operation_lease(&tier_config_mgr, &oi.transitioned_object.tier).await,
-        }
-        .map_err(Error::other)?;
-        let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?;
-        if let Some(expected_operation_id) = expected_operation_id {
-            require_restore_operation_id(oi.user_defined.as_ref(), expected_operation_id)?;
-        }
-        let mut ropts = put_restore_opts(bucket, object, &opts.transition.restore_request, &oi).await?;
-        if let Some(expected_operation_id) = expected_operation_id {
-            rustfs_utils::http::metadata_compat::insert_str(
-                &mut ropts.user_defined,
-                SUFFIX_RESTORE_OPERATION_ID,
-                expected_operation_id.to_string(),
-            );
-        }
-        let mut restore_commit_metadata = if let Some(expected_operation_id) = expected_operation_id {
-            let mut metadata = HashMap::new();
-            metadata.insert(X_AMZ_RESTORE.as_str().to_string(), "ongoing-request=\"false\"".to_string());
-            rustfs_utils::http::metadata_compat::insert_str(
-                &mut metadata,
-                SUFFIX_RESTORE_OPERATION_ID,
-                expected_operation_id.to_string(),
-            );
-            metadata
-        } else {
-            HashMap::new()
-        };
-        if let Some(part_checksums) =
-            rustfs_utils::http::get_consistent_str(&actual_fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
-        {
-            rustfs_utils::http::insert_str(
-                &mut restore_commit_metadata,
-                rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
-                part_checksums.to_string(),
-            );
-        }
-        // Keep the public ECStore capacity admission attached to each local
-        // commit. The tier reads below must remain outside the object write
-        // lock so HEAD/GET do not wait for a slow remote copy-back. Restore
-        // does not hold an outer object write lock while copying bytes, so a
-        // caller-supplied boolean is not transferable lock authority: PUT and
-        // Complete must acquire their own commit-late write locks.
-        ropts.no_lock = false;
-        ropts.expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
-        ropts.bucket_lifecycle_lock_fence = opts.bucket_lifecycle_lock_fence.clone();
-        ropts.namespace_lock_fence = opts.namespace_lock_fence.clone();
-        ropts.object_lock_config_snapshot = opts.object_lock_config_snapshot.clone();
-        ropts.decommission_capacity_admission = opts.decommission_capacity_admission.clone();
-        if oi.parts.len() == 1 {
-            let mut opts = opts.clone();
-            opts.part_number = Some(1);
-            let rs: Option<HTTPRangeSpec> = None;
-            let gr = get_transitioned_object_reader_with_tier_manager(
-                bucket,
-                object,
-                &rs,
-                &HeaderMap::new(),
-                &oi,
-                &opts,
-                &self_.ctx.tier_config_mgr(),
-                self_.ctx.object_encryption_resolver(),
-            )
-            .await;
-            if let Err(err) = gr {
-                return set_restore_header_fn(&mut oi, Some(to_object_err(err.into(), vec![bucket, object]))).await;
+            let mut ropts = put_restore_opts(bucket, object, &opts.transition.restore_request, &oi).await?;
+            if let Some(expected_operation_id) = expected_operation_id {
+                rustfs_utils::http::metadata_compat::insert_str(
+                    &mut ropts.user_defined,
+                    SUFFIX_RESTORE_OPERATION_ID,
+                    expected_operation_id.to_string(),
+                );
             }
-            let gr = gr?;
-            let reader = BufReader::new(gr.stream);
-            let hash_reader = HashReader::from_stream(reader, gr.object_info.size, oi.get_actual_size()?, None, None, false)?;
-            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(),
-                        object: restored_info,
-                        user_agent: "Internal: [Restore-Completed]".to_string(),
-                        host: runtime_sources::default_local_node_name(),
-                        ..Default::default()
-                    });
-                    Ok(())
-                }
-                Err(err) => set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await,
+            let mut restore_commit_metadata = if let Some(expected_operation_id) = expected_operation_id {
+                let mut metadata = HashMap::new();
+                metadata.insert(X_AMZ_RESTORE.as_str().to_string(), "ongoing-request=\"false\"".to_string());
+                rustfs_utils::http::metadata_compat::insert_str(
+                    &mut metadata,
+                    SUFFIX_RESTORE_OPERATION_ID,
+                    expected_operation_id.to_string(),
+                );
+                metadata
+            } else {
+                HashMap::new()
             };
-        }
-
-        let res = self_.clone().new_multipart_upload(bucket, object, &ropts).await?;
-        #[cfg(test)]
-        {
-            *RESTORE_MULTIPART_UPLOAD_ID
-                .lock()
-                .expect("restore multipart upload-id lock must not be poisoned") = Some(res.upload_id.clone());
-        }
-        let mut upload_cleanup = RestoreMultipartUploadCleanup::new(self_.clone(), bucket, object, &res.upload_id);
-        let restore_result: Result<ObjectInfo> = async {
-            let mut uploaded_parts: Vec<CompletePart> = vec![];
-            let parts = Arc::clone(&oi.parts);
-            let mut part_offset: i64 = 0;
-            for part_info in parts.iter() {
-                let mut part_opts = opts.clone();
-                part_opts.part_number = Some(part_info.number);
-                #[cfg(test)]
-                fail_restore_multipart_at(RestoreMultipartFailurePoint::InvalidPartSize)?;
-                if part_info.actual_size <= 0 {
-                    return Err(Error::other(format!("invalid multipart restore part size {}", part_info.actual_size)));
-                }
-                #[cfg(test)]
-                fail_restore_multipart_at(RestoreMultipartFailurePoint::RangeOverflow)?;
-                let part_end = part_offset
-                    .checked_add(part_info.actual_size - 1)
-                    .ok_or_else(|| Error::other("multipart restore part range overflow".to_string()))?;
-                let rs = Some(HTTPRangeSpec {
-                    is_suffix_length: false,
-                    start: part_offset,
-                    end: part_end,
-                });
-                part_offset = part_end
-                    .checked_add(1)
-                    .ok_or_else(|| Error::other("multipart restore part offset overflow".to_string()))?;
-                #[cfg(test)]
-                fail_restore_multipart_at(RestoreMultipartFailurePoint::TierGet)?;
+            if let Some(part_checksums) =
+                rustfs_utils::http::get_consistent_str(&actual_fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS)
+            {
+                rustfs_utils::http::insert_str(
+                    &mut restore_commit_metadata,
+                    rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
+                    part_checksums.to_string(),
+                );
+            }
+            // Keep the public ECStore capacity admission attached to each local
+            // commit. The tier reads below must remain outside the object write
+            // lock so HEAD/GET do not wait for a slow remote copy-back. Restore
+            // does not hold an outer object write lock while copying bytes, so a
+            // caller-supplied boolean is not transferable lock authority: PUT and
+            // Complete must acquire their own commit-late write locks.
+            ropts.no_lock = false;
+            ropts.expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
+            ropts.bucket_lifecycle_lock_fence = opts.bucket_lifecycle_lock_fence.clone();
+            ropts.namespace_lock_fence = opts.namespace_lock_fence.clone();
+            ropts.object_lock_config_snapshot = opts.object_lock_config_snapshot.clone();
+            ropts.decommission_capacity_admission = opts.decommission_capacity_admission.clone();
+            if oi.parts.len() == 1 {
+                let mut opts = opts.clone();
+                opts.part_number = Some(1);
+                let rs: Option<HTTPRangeSpec> = None;
                 let gr = get_transitioned_object_reader_with_tier_manager(
                     bucket,
                     object,
                     &rs,
                     &HeaderMap::new(),
                     &oi,
-                    &part_opts,
+                    &opts,
                     &self_.ctx.tier_config_mgr(),
                     self_.ctx.object_encryption_resolver(),
                 )
                 .await
-                .map_err(StorageError::Io)?;
+                .map_err(|err| to_object_err(err.into(), vec![bucket, object]))?;
                 let reader = BufReader::new(gr.stream);
-                #[cfg(test)]
-                fail_restore_multipart_at(RestoreMultipartFailurePoint::HashReader)?;
-                let hash_reader =
-                    HashReader::from_stream(reader, part_info.actual_size, part_info.actual_size, None, None, false)?;
+                let hash_reader = HashReader::from_stream(reader, gr.object_info.size, oi.get_actual_size()?, None, None, false)?;
                 let mut p_reader = PutObjReader::new(hash_reader);
-                #[cfg(test)]
-                fail_restore_multipart_at(RestoreMultipartFailurePoint::PutPart)?;
-                let p_info = self_
+                let restored_info = self_
                     .clone()
-                    .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &ropts)
-                    .await?;
-                #[cfg(test)]
-                let p_info = if restore_multipart_failure_is(RestoreMultipartFailurePoint::SizeMismatch) {
-                    let mut injected = p_info;
-                    injected.size = 0;
-                    injected
-                } else {
-                    p_info
-                };
-                if p_info.size as i64 != part_info.actual_size {
-                    return Err(Error::other(ObjectApiError::InvalidObjectState(GenericError {
-                        bucket: bucket.to_string(),
-                        object: object.to_string(),
-                        ..Default::default()
-                    })));
-                }
-                uploaded_parts.push(CompletePart {
-                    part_num: p_info.part_num,
-                    etag: p_info.etag,
-                    checksum_crc32: None,
-                    checksum_crc32c: None,
-                    checksum_sha1: None,
-                    checksum_sha256: None,
-                    checksum_crc64nvme: None,
+                    .put_object(bucket, object, &mut p_reader, &ropts)
+                    .await
+                    .map_err(|err| to_object_err(err, vec![bucket, object]))?;
+                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(),
+                    object: restored_info,
+                    user_agent: "Internal: [Restore-Completed]".to_string(),
+                    host: runtime_sources::default_local_node_name(),
+                    ..Default::default()
                 });
+                return Ok(());
             }
+
+            let res = self_.clone().new_multipart_upload(bucket, object, &ropts).await?;
             #[cfg(test)]
-            if restore_multipart_failure_is(RestoreMultipartFailurePoint::Complete) {
-                uploaded_parts
-                    .first_mut()
-                    .expect("multipart restore must contain at least one uploaded part")
-                    .etag = Some("injected-invalid-complete-etag".to_string());
+            {
+                *RESTORE_MULTIPART_UPLOAD_ID
+                    .lock()
+                    .expect("restore multipart upload-id lock must not be poisoned") = Some(res.upload_id.clone());
             }
-            let complete_opts = ObjectOptions {
-                mod_time: oi.mod_time,
-                version_id: oi.version_id.map(|version| version.to_string()),
-                expected_bucket_incarnation_id: opts.expected_bucket_incarnation_id,
-                bucket_lifecycle_lock_fence: opts.bucket_lifecycle_lock_fence.clone(),
-                user_defined: restore_commit_metadata,
-                no_lock: false,
-                decommission_capacity_admission: opts.decommission_capacity_admission.clone(),
-                ..Default::default()
+            let mut upload_cleanup = RestoreMultipartUploadCleanup::new(self_.clone(), bucket, object, &res.upload_id);
+            let restore_result: Result<ObjectInfo> = async {
+                let mut uploaded_parts: Vec<CompletePart> = vec![];
+                let parts = Arc::clone(&oi.parts);
+                let mut part_offset: i64 = 0;
+                for part_info in parts.iter() {
+                    let mut part_opts = opts.clone();
+                    part_opts.part_number = Some(part_info.number);
+                    #[cfg(test)]
+                    fail_restore_multipart_at(RestoreMultipartFailurePoint::InvalidPartSize)?;
+                    if part_info.actual_size <= 0 {
+                        return Err(Error::other(format!("invalid multipart restore part size {}", part_info.actual_size)));
+                    }
+                    #[cfg(test)]
+                    fail_restore_multipart_at(RestoreMultipartFailurePoint::RangeOverflow)?;
+                    let part_end = part_offset
+                        .checked_add(part_info.actual_size - 1)
+                        .ok_or_else(|| Error::other("multipart restore part range overflow".to_string()))?;
+                    let rs = Some(HTTPRangeSpec {
+                        is_suffix_length: false,
+                        start: part_offset,
+                        end: part_end,
+                    });
+                    part_offset = part_end
+                        .checked_add(1)
+                        .ok_or_else(|| Error::other("multipart restore part offset overflow".to_string()))?;
+                    #[cfg(test)]
+                    fail_restore_multipart_at(RestoreMultipartFailurePoint::TierGet)?;
+                    let gr = get_transitioned_object_reader_with_tier_manager(
+                        bucket,
+                        object,
+                        &rs,
+                        &HeaderMap::new(),
+                        &oi,
+                        &part_opts,
+                        &self_.ctx.tier_config_mgr(),
+                        self_.ctx.object_encryption_resolver(),
+                    )
+                    .await
+                    .map_err(StorageError::Io)?;
+                    let reader = BufReader::new(gr.stream);
+                    #[cfg(test)]
+                    fail_restore_multipart_at(RestoreMultipartFailurePoint::HashReader)?;
+                    let hash_reader =
+                        HashReader::from_stream(reader, part_info.actual_size, part_info.actual_size, None, None, false)?;
+                    let mut p_reader = PutObjReader::new(hash_reader);
+                    #[cfg(test)]
+                    fail_restore_multipart_at(RestoreMultipartFailurePoint::PutPart)?;
+                    let p_info = self_
+                        .clone()
+                        .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &ropts)
+                        .await?;
+                    #[cfg(test)]
+                    let p_info = if restore_multipart_failure_is(RestoreMultipartFailurePoint::SizeMismatch) {
+                        let mut injected = p_info;
+                        injected.size = 0;
+                        injected
+                    } else {
+                        p_info
+                    };
+                    if p_info.size as i64 != part_info.actual_size {
+                        return Err(Error::other(ObjectApiError::InvalidObjectState(GenericError {
+                            bucket: bucket.to_string(),
+                            object: object.to_string(),
+                            ..Default::default()
+                        })));
+                    }
+                    uploaded_parts.push(CompletePart {
+                        part_num: p_info.part_num,
+                        etag: p_info.etag,
+                        checksum_crc32: None,
+                        checksum_crc32c: None,
+                        checksum_sha1: None,
+                        checksum_sha256: None,
+                        checksum_crc64nvme: None,
+                    });
+                }
+                #[cfg(test)]
+                if restore_multipart_failure_is(RestoreMultipartFailurePoint::Complete) {
+                    uploaded_parts
+                        .first_mut()
+                        .expect("multipart restore must contain at least one uploaded part")
+                        .etag = Some("injected-invalid-complete-etag".to_string());
+                }
+                let complete_opts = ObjectOptions {
+                    mod_time: oi.mod_time,
+                    version_id: oi.version_id.map(|version| version.to_string()),
+                    expected_bucket_incarnation_id: opts.expected_bucket_incarnation_id,
+                    bucket_lifecycle_lock_fence: opts.bucket_lifecycle_lock_fence.clone(),
+                    user_defined: restore_commit_metadata,
+                    no_lock: false,
+                    decommission_capacity_admission: opts.decommission_capacity_admission.clone(),
+                    ..Default::default()
+                };
+                self_
+                    .clone()
+                    .complete_multipart_upload(bucket, object, &res.upload_id, uploaded_parts, &complete_opts)
+                    .await
+            }
+            .await;
+            let restored_info = match restore_result {
+                Ok(info) => {
+                    upload_cleanup.disarm();
+                    info
+                }
+                Err(err) => {
+                    upload_cleanup.abort().await;
+                    return Err(err);
+                }
             };
-            self_
-                .clone()
-                .complete_multipart_upload(bucket, object, &res.upload_id, uploaded_parts, &complete_opts)
-                .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(),
+                object: restored_info,
+                user_agent: "Internal: [Restore-Completed]".to_string(),
+                host: runtime_sources::default_local_node_name(),
+                ..Default::default()
+            });
+            Ok(())
         }
         .await;
-        let restored_info = match restore_result {
-            Ok(info) => {
-                upload_cleanup.disarm();
-                info
+        if let Err(primary_error) = restore_result {
+            if let Err(cleanup_error) = self_.update_restore_metadata(bucket, object, &oi, opts).await {
+                warn!(
+                    event = EVENT_LIFECYCLE_RESTORE_CLEANUP,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    state = "failed",
+                    bucket,
+                    object,
+                    primary_error = %primary_error,
+                    cleanup_error = %cleanup_error,
+                    "restore metadata cleanup failed"
+                );
             }
-            Err(err) => {
-                upload_cleanup.abort().await;
-                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(),
-            object: restored_info,
-            user_agent: "Internal: [Restore-Completed]".to_string(),
-            host: runtime_sources::default_local_node_name(),
-            ..Default::default()
-        });
+            return Err(primary_error);
+        }
         Ok(())
     }
 
@@ -12819,10 +12858,24 @@ mod transition_commit_failure_tests {
 
     pub(super) fn restore_metadata(operation_id: Uuid, ongoing: bool) -> HashMap<String, String> {
         let mut metadata = restore_operation_id_metadata(operation_id);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_WORKER_LOCK,
+            rustfs_utils::http::metadata_compat::RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string(),
+        );
         metadata.insert(s3s::header::X_AMZ_RESTORE.as_str().to_string(), format!("ongoing-request=\"{ongoing}\""));
         metadata
     }
 
+    #[test]
+    fn restore_tier_mutation_retry_backoff_is_bounded() {
+        assert_eq!(restore_tier_mutation_retry_delay(0), Duration::from_millis(250));
+        assert_eq!(restore_tier_mutation_retry_delay(1), Duration::from_millis(500));
+        assert_eq!(restore_tier_mutation_retry_delay(4), Duration::from_secs(4));
+        assert_eq!(restore_tier_mutation_retry_delay(5), RESTORE_TIER_MUTATION_RETRY_CAP);
+        assert_eq!(restore_tier_mutation_retry_delay(u32::MAX), RESTORE_TIER_MUTATION_RETRY_CAP);
+    }
+
     #[tokio::test]
     async fn rejected_unsupported_remote_versions_are_cleaned_up() {
         for remote_version in ["null", "opaque-version-token"] {
@@ -12961,8 +13014,21 @@ mod transition_commit_failure_tests {
             *RESTORE_MULTIPART_UPLOAD_ID
                 .lock()
                 .expect("restore multipart upload-id lock must not be poisoned") = None;
+            let operation_id = Uuid::new_v4();
+            set_disks
+                .put_object_metadata(
+                    bucket,
+                    object,
+                    &ObjectOptions {
+                        eval_metadata: Some(restore_metadata(operation_id, true)),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("the injected restore generation should be installed");
             let mut opts = ObjectOptions::default();
             opts.transition.restore_request.days = Some(1);
+            opts.user_defined = restore_operation_id_metadata(operation_id);
             set_disks
                 .clone()
                 .restore_transitioned_object(bucket, object, &opts)
@@ -12985,6 +13051,24 @@ mod transition_commit_failure_tests {
                     "{point:?}: failed restore must remove staged multipart data"
                 );
             }
+            let cleaned = set_disks
+                .get_object_info(bucket, object, &ObjectOptions::default())
+                .await
+                .expect("failed restore cleanup should leave the transitioned object readable");
+            assert!(
+                !cleaned.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()),
+                "{point:?}: every post-snapshot failure must clean the public ongoing marker"
+            );
+            assert!(
+                rustfs_utils::http::get_str(cleaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID,)
+                    .is_none(),
+                "{point:?}: every post-snapshot failure must clean its operation generation"
+            );
+            assert!(
+                rustfs_utils::http::get_str(cleaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,)
+                    .is_none(),
+                "{point:?}: every post-snapshot failure must clean its liveness marker"
+            );
         }
         assert_eq!(
             RESTORE_MULTIPART_ABORT_ATTEMPTS.load(Ordering::Relaxed),
@@ -12994,8 +13078,21 @@ mod transition_commit_failure_tests {
         *RESTORE_MULTIPART_FAILURE_POINT
             .lock()
             .expect("restore multipart failure-point lock must not be poisoned") = None;
+        let operation_id = Uuid::new_v4();
+        set_disks
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    eval_metadata: Some(restore_metadata(operation_id, true)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the successful restore generation should be installed");
         let mut opts = ObjectOptions::default();
         opts.transition.restore_request.days = Some(1);
+        opts.user_defined = restore_operation_id_metadata(operation_id);
         set_disks
             .clone()
             .restore_transitioned_object(bucket, object, &opts)
@@ -13043,6 +13140,106 @@ mod transition_commit_failure_tests {
             restore_status.expiry().is_some(),
             "successful multipart restore must retain its expiry date"
         );
+        assert!(
+            rustfs_utils::http::get_str(restored.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,)
+                .is_none(),
+            "successful multipart restore must consume the worker-liveness marker"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn restore_failure_after_snapshot_cleans_exact_generation_and_returns_primary_error() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "restore-post-snapshot-cleanup-bucket";
+        let object = "object.bin";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut reader = PutObjReader::from_vec(b"post-snapshot cleanup source".repeat(1024));
+        let original = set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("source object should be written");
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    transition: TransitionOptions {
+                        status: TRANSITION_PENDING.to_string(),
+                        tier: tier_name,
+                        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()
+                },
+            )
+            .await
+            .expect("source object should transition before restore");
+
+        let operation_id = Uuid::new_v4();
+        let (mut source_fi, _, online_disks) = set_disks
+            .get_object_fileinfo(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    metadata_cache_safe: false,
+                    ..Default::default()
+                },
+                true,
+                false,
+            )
+            .await
+            .expect("transitioned metadata should be readable")
+            .into_owned();
+        source_fi.metadata.extend(restore_metadata(operation_id, true));
+        rustfs_utils::http::insert_str(
+            &mut source_fi.metadata,
+            rustfs_utils::http::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+            "invalid".to_string(),
+        );
+        set_disks
+            .update_object_meta(bucket, object, source_fi, &online_disks)
+            .await
+            .expect("invalid backend identity fixture should be persisted");
+        set_disks.invalidate_get_object_metadata_cache(bucket, object).await;
+
+        let mut opts = ObjectOptions::default();
+        opts.transition.restore_request.days = Some(1);
+        opts.user_defined = restore_operation_id_metadata(operation_id);
+        let error = set_disks
+            .clone()
+            .restore_transitioned_object(bucket, object, &opts)
+            .await
+            .expect_err("invalid backend identity must fail before the tier read");
+        assert!(
+            error
+                .to_string()
+                .contains("transition tier backend identity has an invalid length"),
+            "cleanup must preserve the primary validation error: {error}"
+        );
+
+        let cleaned = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("cleanup should leave the transitioned object readable");
+        assert_eq!(cleaned.transitioned_object.status, TRANSITION_COMPLETE);
+        assert!(!cleaned.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()));
+        assert!(
+            rustfs_utils::http::get_str(cleaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID,)
+                .is_none()
+        );
+        assert!(
+            rustfs_utils::http::get_str(cleaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,).is_none()
+        );
     }
 
     #[tokio::test]
@@ -14267,6 +14464,11 @@ mod transition_commit_failure_tests {
             .is_none(),
             "completed restore PUT must not persist the internal operation id"
         );
+        assert!(
+            rustfs_utils::http::get_str(restored.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,)
+                .is_none(),
+            "completed restore PUT must not persist the worker-liveness marker"
+        );
     }
 
     #[tokio::test]
@@ -14415,6 +14617,11 @@ mod transition_commit_failure_tests {
             .is_none(),
             "completed multipart restore must not persist the internal operation id"
         );
+        assert!(
+            rustfs_utils::http::get_str(restored.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,)
+                .is_none(),
+            "completed multipart restore must not persist the worker-liveness marker"
+        );
     }
 
     #[tokio::test]

crates/ecstore/src/set_disk/replication.rs

@@ -213,6 +213,12 @@ impl SetDisks {
             }
             .to_string(),
         );
+        for suffix in [
+            rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID,
+            rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_WORKER_LOCK,
+        ] {
+            rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, suffix);
+        }
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?;
         if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
@@ -291,11 +297,8 @@ impl SetDisks {
             .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
             .await?
             .into_owned();
-        if let Some(expected_operation_id) = expected_operation_id {
-            match restore_operation_id_from_metadata(&fi.metadata)? {
-                Some(actual_operation_id) if actual_operation_id == expected_operation_id => {}
-                _ => return Ok(()),
-            }
+        if restore_operation_id_from_metadata(&fi.metadata)? != expected_operation_id {
+            return Ok(());
         }
         if !expected.matches_file_info(&fi, &expected_etag) {
             return Ok(());
@@ -308,6 +311,10 @@ impl SetDisks {
             &mut fi.metadata,
             rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID,
         );
+        rustfs_utils::http::metadata_compat::remove_str(
+            &mut fi.metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_WORKER_LOCK,
+        );
         if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
             || decommission_object_lock_guard
                 .as_ref()

crates/ecstore/src/store/object.rs

@@ -71,6 +71,8 @@ use tokio::io::{AsyncRead, ReadBuf};
 const RECURSIVE_DELETE_VERSION_SCAN_PAGE_SIZE: i32 = 1000;
 #[cfg(test)]
 const RECURSIVE_DELETE_VERSION_SCAN_PAGE_SIZE: i32 = 2;
+const RESTORE_WORKER_LOCK_PREFIX: &str = "ilm/restore-worker-locks";
+const RESTORE_WORKER_LOCK_PROBE_TIMEOUT: Duration = Duration::from_millis(50);
 
 fn install_tier_free_version_receipt_sink(opts: &mut ObjectOptions) -> Option<TierFreeVersionReceiptSink> {
     if opts.tier_free_version_receipt_sink.is_some() || opts.skip_free_version || opts.delete_prefix {
@@ -934,6 +936,17 @@ impl RestoreAcceptGuard {
     }
 }
 
+/// Liveness proof for one asynchronous restore generation. Unlike the object
+/// accept/commit locks, this lock lives in the reserved metadata namespace and
+/// never blocks reads or unrelated writes of the restored object.
+pub struct RestoreWorkerGuard(rustfs_lock::NamespaceLockGuard);
+
+impl RestoreWorkerGuard {
+    pub fn is_lock_lost(&self) -> bool {
+        self.0.is_lock_lost()
+    }
+}
+
 impl Drop for ObjectLockDiagGuard {
     fn drop(&mut self) {
         if !self.enabled || self.guard.is_released() {
@@ -2670,6 +2683,32 @@ impl ECStore {
         Ok(RestoreAcceptGuard(guard))
     }
 
+    /// Acquire the liveness lock for a newly generated restore UUID using the
+    /// normal namespace-lock timeout. New generations cannot legitimately
+    /// contend, but a distributed quorum can take longer than the short orphan
+    /// probe window under load.
+    pub async fn acquire_restore_worker_guard(&self, operation_id: Uuid) -> Result<RestoreWorkerGuard> {
+        let object = format!("{RESTORE_WORKER_LOCK_PREFIX}/{operation_id}");
+        let lock = self.handle_new_ns_lock(RUSTFS_META_BUCKET, &object).await?;
+        lock.get_write_lock_quiet(get_lock_acquire_timeout())
+            .await
+            .map(RestoreWorkerGuard)
+            .map_err(|err| Self::map_namespace_lock_error(RUSTFS_META_BUCKET, &object, "restore_worker", err))
+    }
+
+    /// Probe the liveness lock for an already-persisted restore UUID.
+    /// Contention is returned as `None`; infrastructure/quorum failures remain
+    /// typed errors so callers fail closed instead of reaping an active worker.
+    pub async fn try_acquire_restore_worker_guard(&self, operation_id: Uuid) -> Result<Option<RestoreWorkerGuard>> {
+        let object = format!("{RESTORE_WORKER_LOCK_PREFIX}/{operation_id}");
+        let lock = self.handle_new_ns_lock(RUSTFS_META_BUCKET, &object).await?;
+        match lock.get_write_lock_quiet(RESTORE_WORKER_LOCK_PROBE_TIMEOUT).await {
+            Ok(guard) => Ok(Some(RestoreWorkerGuard(guard))),
+            Err(rustfs_lock::LockError::Timeout { .. } | rustfs_lock::LockError::AlreadyLocked { .. }) => Ok(None),
+            Err(err) => Err(Self::map_namespace_lock_error(RUSTFS_META_BUCKET, &object, "restore_worker", err)),
+        }
+    }
+
     async fn acquire_delete_objects_write_locks(
         &self,
         bucket: &str,
@@ -8870,6 +8909,36 @@ mod tests {
         assert!(matches!(err, StorageError::Lock(rustfs_lock::LockError::Timeout { .. })));
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn restore_worker_guard_proves_generation_liveness_until_drop() {
+        let store = Arc::new(new_read_lock_test_store().await);
+        let operation_id = Uuid::from_u128(1);
+
+        let first = store
+            .acquire_restore_worker_guard(operation_id)
+            .await
+            .expect("first worker must own its newly generated lock");
+        assert!(
+            store
+                .try_acquire_restore_worker_guard(operation_id)
+                .await
+                .expect("a contended liveness probe should remain typed")
+                .is_none(),
+            "the same generation must remain live while its worker guard is held"
+        );
+
+        drop(first);
+        assert!(
+            store
+                .try_acquire_restore_worker_guard(operation_id)
+                .await
+                .expect("post-drop liveness probe should not fail")
+                .is_some(),
+            "dropping the worker guard must make the orphan generation recoverable"
+        );
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn reader_lock_is_held_when_optimization_is_disabled() {

crates/filemeta/src/filemeta.rs

@@ -30,7 +30,7 @@ use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, SUFFIX_CRC, SUFFIX_DATA_MOV, SUFFIX_HEALING,
     SUFFIX_PURGESTATUS, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX,
     SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID,
-    contains_key_str, has_internal_suffix, insert_bytes, is_internal_key, remove_bytes,
+    SUFFIX_RESTORE_WORKER_LOCK, contains_key_str, has_internal_suffix, insert_bytes, is_internal_key, remove_bytes,
 };
 use s3s::header::X_AMZ_RESTORE;
 use serde::{Deserialize, Serialize};
@@ -358,8 +358,10 @@ impl FileMeta {
                         if let Some(ref mut obj) = ver.object {
                             if replace_user_metadata {
                                 obj.meta_user.clear();
-                                if !contains_key_str(&fi.metadata, SUFFIX_RESTORE_OPERATION_ID) {
-                                    remove_bytes(&mut obj.meta_sys, SUFFIX_RESTORE_OPERATION_ID);
+                                for suffix in [SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK] {
+                                    if !contains_key_str(&fi.metadata, suffix) {
+                                        remove_bytes(&mut obj.meta_sys, suffix);
+                                    }
                                 }
                             }
 
@@ -1780,6 +1782,12 @@ mod test {
         fi.metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), "1".to_string());
         fi.metadata
             .insert(AMZ_RESTORE_REQUEST_DATE.to_string(), "Thu, 16 Jul 2026 00:00:00 GMT".to_string());
+        rustfs_utils::http::insert_str(&mut fi.metadata, SUFFIX_RESTORE_OPERATION_ID, Uuid::from_u128(1).to_string());
+        rustfs_utils::http::insert_str(
+            &mut fi.metadata,
+            SUFFIX_RESTORE_WORKER_LOCK,
+            rustfs_utils::http::RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string(),
+        );
         fm.add_version(fi).unwrap();
 
         let expire_fi = FileInfo {
@@ -1796,6 +1804,14 @@ mod test {
         assert!(!after.metadata.contains_key(AMZ_RESTORE), "x-amz-restore must be stripped");
         assert!(!after.metadata.contains_key(AMZ_RESTORE_EXPIRY_DAYS));
         assert!(!after.metadata.contains_key(AMZ_RESTORE_REQUEST_DATE));
+        assert!(
+            rustfs_utils::http::get_str(&after.metadata, SUFFIX_RESTORE_OPERATION_ID).is_none(),
+            "expired restore must not retain its operation generation"
+        );
+        assert!(
+            rustfs_utils::http::get_str(&after.metadata, SUFFIX_RESTORE_WORKER_LOCK).is_none(),
+            "expired restore must not retain the worker-liveness protocol marker"
+        );
         assert_eq!(after.transition_status, TRANSITION_COMPLETE);
         assert_eq!(after.transitioned_objname, "remote/obj");
         assert_eq!(after.transition_tier, "COLDTIER");

crates/filemeta/src/filemeta/version.rs

@@ -30,11 +30,12 @@ use crate::{ChecksumInfo, TransitionVersionState};
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::http::{
     RUSTFS_INTERNAL_PREFIX, SUFFIX_CRC, SUFFIX_FREE_VERSION, SUFFIX_INLINE_DATA, SUFFIX_PART_CHECKSUMS, SUFFIX_PURGESTATUS,
-    SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX, SUFFIX_REPLICATION_RESET_ARN_PREFIX, SUFFIX_TIER_FV_ID,
-    SUFFIX_TIER_FV_MARKER, SUFFIX_TRANSITION_STATUS, SUFFIX_TRANSITION_TIER, SUFFIX_TRANSITION_TIER_DESTINATION_ID,
-    SUFFIX_TRANSITIONED_OBJECTNAME, SUFFIX_TRANSITIONED_VERSION_ID, SUFFIX_TRANSITIONED_VERSION_STATE, contains_key_bytes,
-    get_bytes, get_consistent_bytes, get_str, has_internal_suffix, insert_bytes, is_internal_key, remove_bytes,
-    strip_internal_prefix, strip_internal_prefix_preserving_case, target_delete_marker_versions,
+    SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX, SUFFIX_REPLICATION_RESET_ARN_PREFIX, SUFFIX_RESTORE_OPERATION_ID,
+    SUFFIX_RESTORE_WORKER_LOCK, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TRANSITION_STATUS, SUFFIX_TRANSITION_TIER,
+    SUFFIX_TRANSITION_TIER_DESTINATION_ID, SUFFIX_TRANSITIONED_OBJECTNAME, SUFFIX_TRANSITIONED_VERSION_ID,
+    SUFFIX_TRANSITIONED_VERSION_STATE, contains_key_bytes, get_bytes, get_consistent_bytes, get_str, has_internal_suffix,
+    insert_bytes, is_internal_key, remove_bytes, strip_internal_prefix, strip_internal_prefix_preserving_case,
+    target_delete_marker_versions,
 };
 
 const MSGPACK_EXT8: u8 = 0xc7;
@@ -2699,6 +2700,8 @@ impl MetaObject {
         self.meta_user.remove(X_AMZ_RESTORE.as_str());
         self.meta_user.remove(AMZ_RESTORE_EXPIRY_DAYS);
         self.meta_user.remove(AMZ_RESTORE_REQUEST_DATE);
+        remove_bytes(&mut self.meta_sys, SUFFIX_RESTORE_OPERATION_ID);
+        remove_bytes(&mut self.meta_sys, SUFFIX_RESTORE_WORKER_LOCK);
     }
 
     pub fn uses_data_dir(&self) -> bool {

crates/utils/src/http/metadata_compat.rs

@@ -65,6 +65,10 @@ pub const SUFFIX_TRANSITION_TIER: &str = "transition-tier";
 pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destination-id";
 pub const SUFFIX_TRANSITION_TRANSACTION_ID: &str = "transition-transaction-id";
 pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id";
+/// Marks restore generations whose worker owns the matching distributed
+/// liveness lock for the duration of the asynchronous copy-back.
+pub const SUFFIX_RESTORE_WORKER_LOCK: &str = "restore-worker-lock";
+pub const RESTORE_WORKER_LOCK_PROTOCOL_V1: &str = "v1";
 pub const SUFFIX_BUCKET_INCARNATION_ID: &str = "bucket-incarnation-id";
 pub const SUFFIX_OBJECT_TRANSACTION_EPOCH: &str = "object-transaction-epoch";
 /// Active rebalance run id mirrored onto `rebalance.bin` object metadata.

docs/architecture/compat-cleanup-register.md

@@ -11,6 +11,7 @@
 
 ## Open Items
 
+- `backlog-1337` legacy restore orphan recovery: releases that predate the restore worker-lock marker can leave a valid operation-id and `ongoing-request="true"` after cancellation or process failure, with no durable liveness proof. New servers allow an exact, non-nil legacy generation to be superseded only when its consistently parsed request date is at least 24 hours old. Remove the clock-based legacy fallback after the minimum supported direct-upgrade release writes the v1 worker-lock marker on every restore and operators have resolved every retained pre-v1 ongoing generation.
 - `tokio-tar-extension-limits` bounded archive parser hardening: Snowball extraction depends on per-entry and cumulative GNU long-name, GNU long-link, and PAX extension limits; physical-entry, GNU sparse-map, and sparse-continuation limits; cancellation-safe sparse parsing; and fused entry streams after parser errors. The released tokio-tar API does not provide this complete boundary. Keep the reviewed fork pin until astral-sh/tokio-tar#118 is merged and one published tokio-tar release contains every listed capability with the Snowball regression fixtures passing against that release.
 - `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources.
 - `backlog-2122` rc.1-rc.3 non-empty scanner usage floor recovery: leadership fencing in those releases can stamp scanner_epoch onto a real bucket-usage snapshot before any scanner cycle completed, leaving a non-empty floor with no scanner_cycle and no authoritative baseline identity. New scanners recognize only this consistent incomplete fenced shape, preserve the epoch through the CAS-protected recovery marker, and rebuild namespace coverage without treating the old usage data as authoritative. Remove this recovery path after rc.1, rc.2, and rc.3 are no longer supported direct-upgrade sources.

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -2246,6 +2246,114 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
     get_barrier.release();
 }
 
+/// rustfs/backlog#1337: cancellation after the ongoing metadata commit but
+/// before detached worker creation must not strand the object forever. The
+/// replacement POST proves the abandoned v1 generation has no live worker,
+/// atomically supersedes it, and performs exactly one remote copy-back.
+#[serial]
+#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1337"]
+#[test]
+fn restore_object_usecase_recovers_cancelled_post_commit_generation() {
+    std::thread::Builder::new()
+        .name("lifecycle-restore-orphan-recovery".to_string())
+        .stack_size(32 * 1024 * 1024)
+        .spawn(|| {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("orphan restore recovery test runtime should build");
+            runtime.block_on(restore_object_usecase_recovers_cancelled_post_commit_generation_inner());
+        })
+        .expect("orphan restore recovery test thread should spawn")
+        .join()
+        .expect("orphan restore recovery test thread should finish");
+}
+
+async fn restore_object_usecase_recovers_cancelled_post_commit_generation_inner() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+    let backend = register_mock_tier(&tier_name).await;
+    let bucket = format!("test-api-restore-orphan-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/restore/orphaned-generation.bin";
+    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
+
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
+    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
+    backend.clear_op_log().await;
+    let tier_gets_before_restore = backend.get_count().await;
+
+    let restore_input = || {
+        RestoreObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .restore_request(Some(RestoreRequest {
+                days: Some(1),
+                description: None,
+                glacier_job_parameters: None,
+                output_location: None,
+                select_parameters: None,
+                tier: None,
+                type_: None,
+            }))
+            .build()
+            .expect("restore request should build")
+    };
+
+    let commit_barrier = crate::app::object::RestoreStatusCommitBarrier::install(bucket.as_str(), object);
+    let first_input = restore_input();
+    let first = tokio::spawn(async move {
+        DefaultObjectUsecase::from_global()
+            .execute_restore_object(build_request(first_input, Method::POST))
+            .await
+    });
+    commit_barrier.wait_until_paused().await;
+
+    first.abort();
+    assert!(first.await.expect_err("the first request must be cancelled").is_cancelled());
+    drop(commit_barrier);
+
+    let orphaned = ecstore
+        .get_object_info(bucket.as_str(), object, &ObjectOptions::default())
+        .await
+        .expect("the committed orphan generation should remain readable");
+    assert!(orphaned.restore_ongoing, "the interrupted request must have committed ongoing=true");
+    assert_eq!(
+        rustfs_utils::http::get_consistent_str(orphaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,),
+        Some(rustfs_utils::http::RESTORE_WORKER_LOCK_PROTOCOL_V1),
+        "recoverable generations must advertise the worker-lock protocol"
+    );
+    let orphaned_operation_id =
+        rustfs_utils::http::get_consistent_str(orphaned.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID)
+            .and_then(|value| Uuid::parse_str(value).ok())
+            .expect("recoverable generations must persist a non-nil operation id");
+    assert!(!orphaned_operation_id.is_nil());
+
+    DefaultObjectUsecase::from_global()
+        .execute_restore_object(build_request(restore_input(), Method::POST))
+        .await
+        .expect("a new POST must supersede the committed generation whose worker lock was released");
+
+    let completed = wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
+        .await
+        .unwrap_or_else(|err| panic!("{err}"));
+    assert!(!completed.restore_ongoing);
+    assert!(completed.restore_expires.is_some());
+    assert!(
+        rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_OPERATION_ID,).is_none(),
+        "completed replacement restore must consume its operation id"
+    );
+    assert!(
+        rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_RESTORE_WORKER_LOCK,).is_none(),
+        "completed replacement restore must consume its liveness marker"
+    );
+    assert_eq!(
+        backend.get_count().await - tier_gets_before_restore,
+        1,
+        "the cancelled pre-spawn generation must not issue a tier GET"
+    );
+}
+
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#4879"]
 #[test]

rustfs/src/app/object/mod.rs

@@ -145,9 +145,10 @@ use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_K
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE,
-    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICA_STATUS,
-    SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_GENERATION, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP,
-    SUFFIX_RESTORE_OPERATION_ID, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST, get_header,
+    RESTORE_WORKER_LOCK_PROTOCOL_V1, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_PLAINTEXT_CHECKSUM,
+    SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_GENERATION, SUFFIX_REPLICATION_STATUS,
+    SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SUFFIX_SOURCE_REPLICATION_CHECK,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header,
     headers::{
         AMZ_CONTENT_SHA256, AMZ_DECODED_CONTENT_LENGTH, AMZ_MINIO_SNOWBALL_IGNORE_DIRS, AMZ_MINIO_SNOWBALL_IGNORE_ERRORS,
         AMZ_MINIO_SNOWBALL_PREFIX, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE,
@@ -224,6 +225,8 @@ pub(crate) use self::internal_put::*;
 pub(crate) use self::on_demand_migration_put::*;
 use self::put::*;
 pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
+#[cfg(test)]
+pub(crate) use self::restore::RestoreStatusCommitBarrier;
 pub(crate) use self::shared::*;
 #[cfg(test)]
 use self::test_support::*;
@@ -241,7 +244,10 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::{Duration, Instant};
-use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+use time::{
+    OffsetDateTime,
+    format_description::well_known::{Rfc2822, Rfc3339},
+};
 use tokio::io::{AsyncRead, ReadBuf};
 use tokio::sync::{OwnedSemaphorePermit, RwLock};
 use tokio_tar::Archive;

rustfs/src/app/object/restore.rs

@@ -16,6 +16,137 @@
 
 use super::*;
 
+// RUSTFS_COMPAT_TODO(backlog-1337): legacy restores lack a liveness marker. Remove after the minimum supported release writes v1 on every restore.
+const LEGACY_RESTORE_ORPHAN_GRACE: time::Duration = time::Duration::hours(24);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum OngoingRestoreRecovery {
+    ActiveOrUnsafe,
+    ProbeWorker(Uuid),
+    SupersedeLegacy,
+}
+
+fn consistent_metadata_value_case_insensitive<'a>(metadata: &'a HashMap<String, String>, key: &str) -> Option<&'a str> {
+    let mut value = None;
+    for (candidate_key, candidate_value) in metadata {
+        if !candidate_key.eq_ignore_ascii_case(key) {
+            continue;
+        }
+        if candidate_value.is_empty() || value.is_some_and(|current| current != candidate_value) {
+            return None;
+        }
+        value = Some(candidate_value.as_str());
+    }
+    value
+}
+
+fn restore_request_date(metadata: &HashMap<String, String>) -> Option<OffsetDateTime> {
+    let raw = consistent_metadata_value_case_insensitive(metadata, AMZ_RESTORE_REQUEST_DATE)?;
+    OffsetDateTime::parse(raw, &Rfc3339)
+        .or_else(|_| OffsetDateTime::parse(raw, &Rfc2822))
+        .ok()
+}
+
+fn restore_operation_id(metadata: &HashMap<String, String>) -> Option<Uuid> {
+    let raw = get_consistent_str(metadata, SUFFIX_RESTORE_OPERATION_ID)?;
+    Uuid::parse_str(raw).ok().filter(|operation_id| !operation_id.is_nil())
+}
+
+fn classify_ongoing_restore(metadata: &HashMap<String, String>, now: OffsetDateTime) -> OngoingRestoreRecovery {
+    let Some(operation_id) = restore_operation_id(metadata) else {
+        return OngoingRestoreRecovery::ActiveOrUnsafe;
+    };
+
+    match get_consistent_str(metadata, SUFFIX_RESTORE_WORKER_LOCK) {
+        Some(RESTORE_WORKER_LOCK_PROTOCOL_V1) => OngoingRestoreRecovery::ProbeWorker(operation_id),
+        Some(_) => OngoingRestoreRecovery::ActiveOrUnsafe,
+        None if contains_key_str(metadata, SUFFIX_RESTORE_WORKER_LOCK) => OngoingRestoreRecovery::ActiveOrUnsafe,
+        None => {
+            let Some(requested_at) = restore_request_date(metadata) else {
+                return OngoingRestoreRecovery::ActiveOrUnsafe;
+            };
+            if requested_at
+                .checked_add(LEGACY_RESTORE_ORPHAN_GRACE)
+                .is_some_and(|reap_after| now >= reap_after)
+            {
+                OngoingRestoreRecovery::SupersedeLegacy
+            } else {
+                OngoingRestoreRecovery::ActiveOrUnsafe
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+struct RestoreStatusCommitBarrierState {
+    bucket: String,
+    object: String,
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+static RESTORE_STATUS_COMMIT_BARRIER: OnceLock<Mutex<Option<Arc<RestoreStatusCommitBarrierState>>>> = OnceLock::new();
+
+#[cfg(test)]
+pub(crate) struct RestoreStatusCommitBarrier {
+    state: Arc<RestoreStatusCommitBarrierState>,
+}
+
+#[cfg(test)]
+impl RestoreStatusCommitBarrier {
+    pub(crate) fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(RestoreStatusCommitBarrierState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            arrived: tokio::sync::Notify::new(),
+            release: tokio::sync::Notify::new(),
+        });
+        let mut slot = RESTORE_STATUS_COMMIT_BARRIER
+            .get_or_init(|| Mutex::new(None))
+            .lock()
+            .expect("restore status commit barrier mutex should not poison");
+        assert!(slot.is_none(), "restore status commit barrier must be installed by one test at a time");
+        *slot = Some(Arc::clone(&state));
+        Self { state }
+    }
+
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
+            .await
+            .expect("restore accept should reach the post-commit barrier");
+    }
+}
+
+#[cfg(test)]
+impl Drop for RestoreStatusCommitBarrier {
+    fn drop(&mut self) {
+        let mut slot = RESTORE_STATUS_COMMIT_BARRIER
+            .get_or_init(|| Mutex::new(None))
+            .lock()
+            .expect("restore status commit barrier mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+        self.state.release.notify_waiters();
+    }
+}
+
+#[cfg(test)]
+async fn maybe_pause_after_restore_status_commit(bucket: &str, object: &str) {
+    let state = RESTORE_STATUS_COMMIT_BARRIER
+        .get_or_init(|| Mutex::new(None))
+        .lock()
+        .expect("restore status commit barrier mutex should not poison")
+        .as_ref()
+        .filter(|state| state.bucket == bucket && state.object == object)
+        .cloned();
+    if let Some(state) = state {
+        state.arrived.notify_one();
+        state.release.notified().await;
+    }
+}
+
 impl DefaultObjectUsecase {
     #[instrument(level = "debug", skip(self, req))]
     pub async fn execute_restore_object(&self, req: S3Request<RestoreObjectInput>) -> S3Result<S3Response<RestoreObjectOutput>> {
@@ -65,6 +196,17 @@ impl DefaultObjectUsecase {
         // write below, so the accept guard would protect nothing for them —
         // they keep the plain (read-locked) accept path.
         let is_select = rreq.type_.as_ref().is_some_and(|t| t.as_str() == "SELECT");
+        let restore_operation_id = (!is_select).then(Uuid::new_v4);
+        let mut restore_worker_guard = if let Some(operation_id) = restore_operation_id {
+            Some(
+                store
+                    .acquire_restore_worker_guard(operation_id)
+                    .await
+                    .map_err(|_| S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."))?,
+            )
+        } else {
+            None
+        };
 
         // Hold the restore-accept guard across the restore-status read, the
         // ongoing/already-restored decision, and the metadata write below, so
@@ -76,11 +218,11 @@ impl DefaultObjectUsecase {
         // in-flight commit on the same object) is transient — answer 503
         // SlowDown so SDK clients back off and retry instead of treating it
         // as a hard failure.
-        let restore_bucket_lifecycle_guard = Some(acquire_copy_bucket_lifecycle_lock(store.as_ref(), &bucket).await?);
+        let mut restore_bucket_lifecycle_guard = Some(acquire_copy_bucket_lifecycle_lock(store.as_ref(), &bucket).await?);
         if store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)? != restore_bucket_incarnation_id {
             return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
         }
-        let accept_guard = if is_select {
+        let mut accept_guard = if is_select {
             None
         } else {
             let guard = store
@@ -112,14 +254,64 @@ impl DefaultObjectUsecase {
             ));
         }
 
-        // Check if restore is already in progress. AWS answers this with
-        // 409 RestoreAlreadyInProgress; a Custom code would serialize as a
-        // retryable 500 and make SDK clients retry the conflict (backlog#1304).
+        // A v1 generation owns a distributed worker-liveness lock. Probe that
+        // lock only after releasing the object/bucket guards: the worker holds
+        // worker-lock -> object-commit-lock, so probing in the opposite order
+        // would create an ABBA cycle. If the probe succeeds, reacquire and
+        // re-read the object before replacing the exact orphan generation.
+        let mut superseded_worker_guard = None;
         if obj_info.restore_ongoing && !is_select {
-            return Err(S3Error::with_message(
-                S3ErrorCode::RestoreAlreadyInProgress,
-                "Object restore is already in progress.",
-            ));
+            match classify_ongoing_restore(obj_info.user_defined.as_ref(), OffsetDateTime::now_utc()) {
+                OngoingRestoreRecovery::ActiveOrUnsafe => {
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::RestoreAlreadyInProgress,
+                        "Object restore is already in progress.",
+                    ));
+                }
+                OngoingRestoreRecovery::SupersedeLegacy => {}
+                OngoingRestoreRecovery::ProbeWorker(previous_operation_id) => {
+                    drop(accept_guard.take());
+                    drop(restore_bucket_lifecycle_guard.take());
+                    let previous_worker_guard = store
+                        .try_acquire_restore_worker_guard(previous_operation_id)
+                        .await
+                        .map_err(|_| S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."))?
+                        .ok_or_else(|| {
+                            S3Error::with_message(S3ErrorCode::RestoreAlreadyInProgress, "Object restore is already in progress.")
+                        })?;
+
+                    restore_bucket_lifecycle_guard = Some(acquire_copy_bucket_lifecycle_lock(store.as_ref(), &bucket).await?);
+                    if store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)?
+                        != restore_bucket_incarnation_id
+                    {
+                        return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
+                    }
+                    accept_guard = Some(
+                        store
+                            .acquire_restore_accept_guard(&bucket, &object)
+                            .await
+                            .map_err(|_| S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."))?,
+                    );
+                    opts.no_lock = true;
+                    obj_info = store.get_object_info(&bucket, &object, &opts).await.map_err(|_| {
+                        S3Error::with_message(S3ErrorCode::Custom("ErrInvalidObjectState".into()), "restore object failed.")
+                    })?;
+                    if obj_info.transitioned_object.status != lifecycle::TRANSITION_COMPLETE {
+                        return Err(S3Error::with_message(
+                            S3ErrorCode::Custom("ErrInvalidTransitionedState".into()),
+                            "restore object failed.",
+                        ));
+                    }
+                    if obj_info.restore_ongoing {
+                        if classify_ongoing_restore(obj_info.user_defined.as_ref(), OffsetDateTime::now_utc())
+                            != OngoingRestoreRecovery::ProbeWorker(previous_operation_id)
+                        {
+                            return Err(S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."));
+                        }
+                        superseded_worker_guard = Some(previous_worker_guard);
+                    }
+                }
+            }
         }
 
         let mut already_restored = false;
@@ -132,7 +324,8 @@ impl DefaultObjectUsecase {
 
         let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), *rreq.days.as_ref().unwrap_or(&1));
         let mut metadata = (*obj_info.user_defined).clone();
-        let restore_operation_id = (!is_select && !already_restored).then(Uuid::new_v4);
+        remove_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID);
+        remove_str(&mut metadata, SUFFIX_RESTORE_WORKER_LOCK);
 
         let mut header = HeaderMap::new();
 
@@ -165,6 +358,7 @@ impl DefaultObjectUsecase {
                 );
                 if let Some(id) = restore_operation_id {
                     insert_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string());
+                    insert_str(&mut metadata, SUFFIX_RESTORE_WORKER_LOCK, RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string());
                 }
             }
             obj_info.user_defined = Arc::new(metadata);
@@ -173,7 +367,9 @@ impl DefaultObjectUsecase {
             // (lock-service degradation), another node may have concurrently
             // accepted this restore — back off instead of committing a second
             // ongoing flag and double-starting the copy-back.
-            if accept_guard.as_ref().is_some_and(|g| g.is_lock_lost()) {
+            if accept_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                || restore_worker_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+            {
                 return Err(S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."));
             }
 
@@ -209,6 +405,9 @@ impl DefaultObjectUsecase {
                 .await
                 .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrCopyObject".into()), "restore object failed."))?;
             rustfs_scanner::record_dirty_usage_bucket(&bucket);
+            #[cfg(test)]
+            maybe_pause_after_restore_status_commit(&bucket, &object).await;
+            drop(superseded_worker_guard.take());
 
             if already_restored {
                 let output = RestoreObjectOutput {
@@ -262,7 +461,9 @@ impl DefaultObjectUsecase {
             insert_str(&mut restore_operation_metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string());
         }
 
+        let restore_worker_guard = restore_worker_guard.take();
         spawn_traced(async move {
+            let _restore_worker_guard = restore_worker_guard;
             let opts = ObjectOptions {
                 transition: TransitionOptions {
                     restore_request: rreq_clone,
@@ -310,6 +511,124 @@ mod tests {
     use http::Method;
     use s3s::dto::RestoreRequest;
 
+    fn ongoing_metadata(operation_id: Uuid) -> HashMap<String, String> {
+        let mut metadata = HashMap::new();
+        insert_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID, operation_id.to_string());
+        metadata
+    }
+
+    #[test]
+    fn ongoing_restore_v1_requires_consistent_nonempty_protocol_and_generation() {
+        let operation_id = Uuid::from_u128(1);
+        let now = OffsetDateTime::parse("2026-02-02T00:00:00Z", &Rfc3339).unwrap();
+
+        let mut dual = ongoing_metadata(operation_id);
+        insert_str(&mut dual, SUFFIX_RESTORE_WORKER_LOCK, RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string());
+        assert_eq!(classify_ongoing_restore(&dual, now), OngoingRestoreRecovery::ProbeWorker(operation_id));
+
+        let mut single = HashMap::new();
+        single.insert(
+            rustfs_utils::http::internal_key_rustfs(SUFFIX_RESTORE_OPERATION_ID),
+            operation_id.to_string(),
+        );
+        single.insert(
+            rustfs_utils::http::internal_key_rustfs(SUFFIX_RESTORE_WORKER_LOCK),
+            RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string(),
+        );
+        assert_eq!(classify_ongoing_restore(&single, now), OngoingRestoreRecovery::ProbeWorker(operation_id));
+
+        let mut minio_only = HashMap::new();
+        minio_only.insert(
+            format!("{}{}", rustfs_utils::http::MINIO_INTERNAL_PREFIX, SUFFIX_RESTORE_OPERATION_ID),
+            operation_id.to_string(),
+        );
+        minio_only.insert(
+            format!("{}{}", rustfs_utils::http::MINIO_INTERNAL_PREFIX, SUFFIX_RESTORE_WORKER_LOCK),
+            RESTORE_WORKER_LOCK_PROTOCOL_V1.to_string(),
+        );
+        assert_eq!(
+            classify_ongoing_restore(&minio_only, now),
+            OngoingRestoreRecovery::ProbeWorker(operation_id)
+        );
+
+        let mut unknown_protocol = dual.clone();
+        insert_str(&mut unknown_protocol, SUFFIX_RESTORE_WORKER_LOCK, "v2".to_string());
+        assert_eq!(classify_ongoing_restore(&unknown_protocol, now), OngoingRestoreRecovery::ActiveOrUnsafe);
+
+        let mut empty_protocol = dual.clone();
+        insert_str(&mut empty_protocol, SUFFIX_RESTORE_WORKER_LOCK, String::new());
+        assert_eq!(classify_ongoing_restore(&empty_protocol, now), OngoingRestoreRecovery::ActiveOrUnsafe);
+
+        let mut conflicting_protocol = dual.clone();
+        conflicting_protocol.insert(
+            format!("{}{}", rustfs_utils::http::MINIO_INTERNAL_PREFIX, SUFFIX_RESTORE_WORKER_LOCK),
+            "v2".to_string(),
+        );
+        assert_eq!(
+            classify_ongoing_restore(&conflicting_protocol, now),
+            OngoingRestoreRecovery::ActiveOrUnsafe
+        );
+
+        for invalid_generation in [String::new(), Uuid::nil().to_string(), "not-a-uuid".to_string()] {
+            let mut metadata = dual.clone();
+            insert_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID, invalid_generation);
+            assert_eq!(classify_ongoing_restore(&metadata, now), OngoingRestoreRecovery::ActiveOrUnsafe);
+        }
+
+        let mut conflicting_generation = dual;
+        conflicting_generation.insert(
+            format!("{}{}", rustfs_utils::http::MINIO_INTERNAL_PREFIX, SUFFIX_RESTORE_OPERATION_ID),
+            Uuid::from_u128(2).to_string(),
+        );
+        assert_eq!(
+            classify_ongoing_restore(&conflicting_generation, now),
+            OngoingRestoreRecovery::ActiveOrUnsafe
+        );
+    }
+
+    #[test]
+    fn legacy_ongoing_restore_is_superseded_only_after_a_valid_stale_request_date() {
+        let operation_id = Uuid::from_u128(1);
+        let now = OffsetDateTime::parse("2026-02-02T00:00:00Z", &Rfc3339).unwrap();
+
+        for stale_date in [
+            "2026-02-01T00:00:00Z",
+            "2026-01-31T23:59:59Z",
+            "Sun, 1 Feb 2026 00:00:00 GMT",
+            "Sat, 31 Jan 2026 23:59:59 GMT",
+        ] {
+            let mut metadata = ongoing_metadata(operation_id);
+            metadata.insert(AMZ_RESTORE_REQUEST_DATE.to_string(), stale_date.to_string());
+            assert_eq!(
+                classify_ongoing_restore(&metadata, now),
+                OngoingRestoreRecovery::SupersedeLegacy,
+                "legacy date {stale_date} should be stale"
+            );
+        }
+
+        for unsafe_date in [
+            None,
+            Some("2026-02-01T00:00:01Z"),
+            Some("2026-02-03T00:00:00Z"),
+            Some("invalid"),
+        ] {
+            let mut metadata = ongoing_metadata(operation_id);
+            if let Some(date) = unsafe_date {
+                metadata.insert(AMZ_RESTORE_REQUEST_DATE.to_string(), date.to_string());
+            }
+            assert_eq!(
+                classify_ongoing_restore(&metadata, now),
+                OngoingRestoreRecovery::ActiveOrUnsafe,
+                "legacy date {unsafe_date:?} must fail closed"
+            );
+        }
+
+        let mut conflicting_date = ongoing_metadata(operation_id);
+        conflicting_date.insert(AMZ_RESTORE_REQUEST_DATE.to_string(), "2026-01-01T00:00:00Z".to_string());
+        conflicting_date.insert(AMZ_RESTORE_REQUEST_DATE.to_ascii_lowercase(), "2025-01-01T00:00:00Z".to_string());
+        assert_eq!(classify_ongoing_restore(&conflicting_date, now), OngoingRestoreRecovery::ActiveOrUnsafe);
+    }
+
     #[tokio::test]
     async fn execute_restore_object_rejects_missing_restore_request() {
         let input = RestoreObjectInput::builder()