diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 0ff7edc8b..89ccf5166 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/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()); } diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 687e1f3b1..377f6ba06 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/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>) -> 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, } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 9e0edd9b5..a258c8fb2 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index f31a7836a..d7d3a39ad 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/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( diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index a68849af7..2b4785f00 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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, 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>, + tier_name: &str, + expected_backend_identity: Option, +) -> Result { + 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| -> 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 = 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 = async { - let mut uploaded_parts: Vec = 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 = 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 = async { + let mut uploaded_parts: Vec = 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 { 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] diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index 0ba4ea2f0..65b9f8a9a 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/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() diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 8d89063f4..4ebcf73be 100644 --- a/crates/ecstore/src/store/object.rs +++ b/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 { 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 { + 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> { + 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() { diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 0e42b1850..70bd66c13 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/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"); diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index 539456e5e..f0969c3cf 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/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 { diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index dbe41e785..f68889468 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/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. diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index cf74e055c..1e10a226c 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/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. diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 53c6fc5f1..cc31c0d9e 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/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 = (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] diff --git a/rustfs/src/app/object/mod.rs b/rustfs/src/app/object/mod.rs index 74cb3e12d..8f9a9955a 100644 --- a/rustfs/src/app/object/mod.rs +++ b/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; diff --git a/rustfs/src/app/object/restore.rs b/rustfs/src/app/object/restore.rs index 8246cb90a..56319265d 100644 --- a/rustfs/src/app/object/restore.rs +++ b/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, 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) -> Option { + 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) -> Option { + 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, 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>>> = OnceLock::new(); + +#[cfg(test)] +pub(crate) struct RestoreStatusCommitBarrier { + state: Arc, +} + +#[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) -> S3Result> { @@ -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 { + 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()