From 9e4c5e949f82d16739c83a3dd1a9e81a549e374a Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 20 Jul 2026 18:32:02 +0800 Subject: [PATCH] fix(ecstore): fence restore cleanup by operation id (#5058) Refs rustfs/backlog#1356 Co-authored-by: heihutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 1 + crates/ecstore/src/set_disk/mod.rs | 25 +++- crates/ecstore/src/set_disk/ops/object.rs | 138 ++++++++++++++++++ crates/ecstore/src/set_disk/replication.rs | 23 ++- crates/filemeta/src/filemeta.rs | 6 +- crates/utils/src/http/metadata_compat.rs | 1 + rustfs/src/app/object_usecase.rs | 10 ++ 7 files changed, 200 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 46ffb068e..3a7916504 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2682,6 +2682,7 @@ pub async fn put_restore_opts( for (k, v) in oi.user_defined.iter() { 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); if !oi.user_tags.is_empty() { meta.insert(AMZ_OBJECT_TAGGING.to_string(), (*oi.user_tags).clone()); } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index ef9b3c955..ee9d44a86 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -150,8 +150,8 @@ use rustfs_utils::http::headers::{ }; use rustfs_utils::http::{ SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, - SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, contains_key_str, get_header_map, get_str, - insert_str, is_encryption_metadata_key, remove_header_map, + SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, contains_key_str, + get_header_map, get_str, insert_str, is_encryption_metadata_key, remove_header_map, }; use rustfs_utils::{ HashAlgorithm, @@ -189,6 +189,27 @@ use tracing::error; use tracing::{Instrument, debug, info, warn}; use uuid::Uuid; +pub(super) fn restore_operation_id_from_metadata(metadata: &HashMap) -> Result> { + let Some(value) = rustfs_utils::http::metadata_compat::get_consistent_str(metadata, SUFFIX_RESTORE_OPERATION_ID) else { + if rustfs_utils::http::metadata_compat::contains_key_str(metadata, SUFFIX_RESTORE_OPERATION_ID) { + return Err(Error::other("invalid restore operation id metadata".to_string())); + } + return Ok(None); + }; + let id = Uuid::parse_str(value).map_err(|_| Error::other("invalid restore operation id metadata".to_string()))?; + if id.is_nil() { + return Err(Error::other("invalid restore operation id metadata".to_string())); + } + Ok(Some(id)) +} + +pub(super) fn require_restore_operation_id(metadata: &HashMap, expected: Uuid) -> Result<()> { + match restore_operation_id_from_metadata(metadata)? { + Some(actual) if actual == expected => Ok(()), + _ => Err(Error::other("restore operation id changed before copy-back".to_string())), + } +} + type ListObjectsV2Info = StorageListObjectsV2Info; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ObjectInfoOrErr = StorageObjectInfoOrErr; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index b92db1b0b..d00c42f1c 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3108,6 +3108,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let (actual_fi, _, _) = fi?; oi = ObjectInfo::from_file_info(&actual_fi, bucket, object, opts.versioned || opts.version_suspended); + if let Some(expected_operation_id) = restore_operation_id_from_metadata(&opts.user_defined)? { + 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?; // The restore copy-back re-writes this same object via put_object / // new_multipart_upload / complete_multipart_upload, each of which takes @@ -4101,6 +4104,141 @@ mod transition_commit_failure_tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn failed_restore_cleanup_ignores_replaced_operation_id_with_same_identity() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-cleanup-operation-id-bucket"; + let object = "object.bin"; + let payload = b"same identity restore cleanup must respect operation id".repeat(1024); + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let operation_a = Uuid::new_v4(); + let operation_b = Uuid::new_v4(); + let mut metadata = HashMap::new(); + metadata.insert(s3s::header::X_AMZ_RESTORE.as_str().to_string(), "ongoing-request=\"true\"".to_string()); + metadata.insert(rustfs_utils::http::headers::AMZ_RESTORE_EXPIRY_DAYS.to_string(), "1".to_string()); + metadata.insert( + rustfs_utils::http::headers::AMZ_RESTORE_REQUEST_DATE.to_string(), + "2026-07-20T00:00:00Z".to_string(), + ); + rustfs_utils::http::metadata_compat::insert_str( + &mut metadata, + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + operation_a.to_string(), + ); + + let mut reader = PutObjReader::from_vec(payload); + set_disks + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + user_defined: metadata, + ..Default::default() + }, + ) + .await + .expect("source object with restore operation A should be written"); + let stale_operation_a = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("operation A metadata should resolve"); + + let mut operation_b_metadata = HashMap::new(); + rustfs_utils::http::metadata_compat::insert_str( + &mut operation_b_metadata, + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + operation_b.to_string(), + ); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(operation_b_metadata), + ..Default::default() + }, + ) + .await + .expect("same-identity restore operation B should replace A"); + + let mut expected_operation_a = HashMap::new(); + rustfs_utils::http::metadata_compat::insert_str( + &mut expected_operation_a, + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + operation_a.to_string(), + ); + set_disks + .update_restore_metadata( + bucket, + object, + &stale_operation_a, + &ObjectOptions { + user_defined: expected_operation_a, + ..Default::default() + }, + ) + .await + .expect("stale operation A cleanup should no-op, not fail"); + let current_operation_b = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("operation B metadata should remain readable"); + assert!( + current_operation_b + .user_defined + .contains_key(s3s::header::X_AMZ_RESTORE.as_str()), + "stale cleanup for operation A must not remove operation B's restore header" + ); + assert_eq!( + rustfs_utils::http::metadata_compat::get_consistent_str( + current_operation_b.user_defined.as_ref(), + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ), + Some(operation_b.to_string().as_str()), + "stale cleanup for operation A must not remove or rewrite operation B" + ); + + let mut expected_operation_b = HashMap::new(); + rustfs_utils::http::metadata_compat::insert_str( + &mut expected_operation_b, + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + operation_b.to_string(), + ); + set_disks + .update_restore_metadata( + bucket, + object, + ¤t_operation_b, + &ObjectOptions { + user_defined: expected_operation_b, + ..Default::default() + }, + ) + .await + .expect("matching operation B cleanup should remove restore markers"); + let cleaned = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("cleaned object metadata should remain readable"); + assert!( + !cleaned.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()), + "matching cleanup must remove the restore header" + ); + assert!( + rustfs_utils::http::metadata_compat::get_consistent_str( + cleaned.user_defined.as_ref(), + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ) + .is_none(), + "matching cleanup must remove the restore operation id" + ); + } + #[tokio::test] #[serial_test::serial] async fn legacy_reload_rejects_route_change_after_local_transition_commit() { diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index 452f0ced3..611abeb7a 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -54,6 +54,7 @@ impl SetDisks { return Ok(()); } let expected = RestoreCleanupIdentity::from_object_info(obj_info); + let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?; let expected_etag = obj_info .etag .clone() @@ -76,17 +77,37 @@ impl SetDisks { let (mut fi, _, disks) = self .get_object_fileinfo_gated(bucket, object, &read_opts, false, false) .await?; + if let Some(expected_operation_id) = expected_operation_id { + match restore_operation_id_from_metadata(&fi.metadata)? { + Some(actual_operation_id) if actual_operation_id == expected_operation_id => {} + _ => return Ok(()), + } + } if !expected.matches_file_info(&fi, &expected_etag) { return Ok(()); } fi.metadata.remove(X_AMZ_RESTORE.as_str()); fi.metadata.remove(AMZ_RESTORE_EXPIRY_DAYS); fi.metadata.remove(AMZ_RESTORE_REQUEST_DATE); + rustfs_utils::http::metadata_compat::remove_str( + &mut fi.metadata, + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ); if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { return Err(Error::other("restore cleanup lock lost before metadata update".to_string())); } self.invalidate_get_object_metadata_cache(bucket, object).await; - self.update_object_meta(bucket, object, fi, disks.as_slice()).await?; + self.update_object_meta_with_opts( + bucket, + object, + fi, + disks.as_slice(), + &UpdateMetadataOpts { + replace_user_metadata: true, + ..Default::default() + }, + ) + .await?; self.invalidate_get_object_metadata_cache(bucket, object).await; Ok(()) } diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 74f85dea4..d292cd9ea 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -26,7 +26,8 @@ use rustfs_utils::http::headers::{ 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_RESET, SUFFIX_REPLICATION_STATUS, - SUFFIX_REPLICATION_TIMESTAMP, has_internal_suffix, insert_bytes, is_internal_key, + SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID, contains_key_str, has_internal_suffix, insert_bytes, + is_internal_key, remove_bytes, }; use s3s::header::X_AMZ_RESTORE; use serde::{Deserialize, Serialize}; @@ -248,6 +249,9 @@ 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 (k, v) in fi.metadata.iter() { diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index 404df9d16..9dc6ba175 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -39,6 +39,7 @@ pub const SUFFIX_TRANSITIONED_OBJECTNAME: &str = "transitioned-object"; pub const SUFFIX_TRANSITIONED_VERSION_ID: &str = "transitioned-versionID"; pub const SUFFIX_TRANSITION_TIER: &str = "transition-tier"; pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destination-id"; +pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id"; pub const SUFFIX_FREE_VERSION: &str = "free-version"; pub const SUFFIX_PURGESTATUS: &str = "purgestatus"; pub const SUFFIX_REPLICA_STATUS: &str = "replica-status"; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 3d3537595..6d3fc37d2 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -137,6 +137,7 @@ use rustfs_utils::CompressionAlgorithm; 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_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, + SUFFIX_RESTORE_OPERATION_ID, 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, @@ -7264,6 +7265,7 @@ 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); let mut header = HeaderMap::new(); @@ -7294,6 +7296,9 @@ impl DefaultObjectUsecase { } .to_string(), ); + if let Some(id) = restore_operation_id { + insert_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string()); + } } obj_info.user_defined = Arc::new(metadata); @@ -7371,6 +7376,10 @@ impl DefaultObjectUsecase { let object_clone = object.clone(); let rreq_clone = rreq.clone(); let version_id_clone = obj_info_.version_id.map(|v| v.to_string()); + let mut restore_operation_metadata = HashMap::new(); + if let Some(id) = restore_operation_id { + insert_str(&mut restore_operation_metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string()); + } spawn_traced(async move { let opts = ObjectOptions { @@ -7380,6 +7389,7 @@ impl DefaultObjectUsecase { ..Default::default() }, version_id: version_id_clone, + user_defined: restore_operation_metadata, ..Default::default() };