From f52dde87d1de14904796fb48f7c89ff192a56dae Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 25 Jul 2026 19:26:50 +0800 Subject: [PATCH] fix(object): make version undo preconditions atomic (#5225) * fix(object): make version undo preconditions atomic * fix(object): fence metadata-only undo copies * fix(object): order versions by commit time --- crates/ecstore/src/object_api/types.rs | 2 + crates/ecstore/src/set_disk/ops/object.rs | 98 +++++++- crates/ecstore/src/store/object.rs | 14 +- docs/operations/atomic-object-undo.md | 41 ++++ .../src/app/lifecycle_transition_api_test.rs | 224 ++++++++++++++++++ rustfs/src/app/object_usecase.rs | 122 +++++++++- rustfs/src/app/storage_api.rs | 4 + 7 files changed, 493 insertions(+), 12 deletions(-) create mode 100644 docs/operations/atomic-object-undo.md diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index a53cc16ae..8fd6dcadd 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -30,6 +30,8 @@ pub struct ObjectOptions { pub delete_prefix: bool, pub delete_prefix_object: bool, pub version_id: Option, + /// RustFS-only compare-and-set condition checked under the object write lock. + pub expected_current_version_id: Option, pub no_lock: bool, /// True when an upper layer already holds the object read lock before /// forwarding a no_lock read to the set layer. diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index a3c22967b..3531a33d8 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1001,11 +1001,7 @@ impl SetDisks { let _ = user_defined.remove(AMZ_STORAGE_CLASS); } - let mod_time = if let Some(mod_time) = opts.mod_time { - Some(mod_time) - } else { - Some(OffsetDateTime::now_utc()) - }; + let mod_time = opts.mod_time; // Drop any disk whose shard did not fully commit (offline at writer // setup, short write, or a write/shutdown error) so its truncated or @@ -1075,6 +1071,45 @@ impl SetDisks { object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } + // Generate ordinary PUT timestamps under the commit lock so version + // ordering follows durable commit ordering when writers queued on + // the same object. Internal callers with an explicit timestamp keep + // their supplied value. + if opts.mod_time.is_none() { + let commit_time = Some(OffsetDateTime::now_utc()); + for pfi in &mut parts_metadatas { + pfi.mod_time = commit_time; + for part in &mut pfi.parts { + part.mod_time = commit_time; + } + } + } + + if let Some(expected) = opts.expected_current_version_id.as_deref() { + let current = self + .get_object_info( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + versioned: true, + ..Default::default() + }, + ) + .await + .map_err(|err| { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + StorageError::PreconditionFailed + } else { + err + } + })?; + if current.version_id.map(|version| version.to_string()).as_deref() != Some(expected) { + return Err(StorageError::PreconditionFailed); + } + } + // Phase 2 (backlog#899): fence the commit on lock loss. If the refresh // heartbeat has observed a refresh-quorum loss, another writer may have // re-acquired this object's lock; committing now would race a double-write. @@ -2323,6 +2358,31 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { ) }; + if let Some(expected) = dst_opts.expected_current_version_id.as_deref() { + let current = self + .get_object_info( + dst_bucket, + dst_object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + versioned: true, + ..Default::default() + }, + ) + .await + .map_err(|err| { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + StorageError::PreconditionFailed + } else { + err + } + })?; + if current.version_id.map(|version| version.to_string()).as_deref() != Some(expected) { + return Err(StorageError::PreconditionFailed); + } + } + self.invalidate_get_object_metadata_cache(dst_bucket, dst_object).await; if dst_opts.http_preconditions.is_some() @@ -2967,6 +3027,34 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { return Ok(ObjectInfo::default()); } + if let Some(expected) = opts.expected_current_version_id.as_deref() { + let current = self + .get_object_info( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + versioned: true, + ..Default::default() + }, + ) + .await + .map_err(|err| { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { + StorageError::PreconditionFailed + } else { + err + } + })?; + if !current.delete_marker + || current.version_id.map(|version| version.to_string()).as_deref() != Some(expected) + || opts.version_id.as_deref() != Some(expected) + { + return Err(StorageError::PreconditionFailed); + } + } + // TODO: Lifecycle let mut version_found = true; diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 41c5ad0c9..2a8d30ee4 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -933,7 +933,7 @@ impl ECStore { let cp_src_dst_same = path_join_buf(&[src_bucket, &src_object]) == path_join_buf(&[dst_bucket, &dst_object]); let mut dst_opts = dst_opts.clone(); - let _dst_lock_guard = if cp_src_dst_same { + let _dst_lock_guard = if cp_src_dst_same && dst_opts.expected_current_version_id.is_none() { self.acquire_object_write_lock_if_needed("copy_object", dst_bucket, &dst_object, &mut dst_opts) .await? } else { @@ -969,6 +969,7 @@ impl ECStore { no_lock: dst_opts.no_lock, mod_time: dst_opts.mod_time, http_preconditions: dst_opts.http_preconditions.clone(), + expected_current_version_id: dst_opts.expected_current_version_id.clone(), ..Default::default() }; return if let Some(reader) = src_info.put_object_reader.as_mut() { @@ -998,6 +999,7 @@ impl ECStore { no_lock: dst_opts.no_lock, mod_time: dst_opts.mod_time, http_preconditions: dst_opts.http_preconditions.clone(), + expected_current_version_id: dst_opts.expected_current_version_id.clone(), ..Default::default() }; return self.pools[pool_idx] @@ -1024,6 +1026,7 @@ impl ECStore { no_lock: dst_opts.no_lock, mod_time: dst_opts.mod_time, http_preconditions: dst_opts.http_preconditions.clone(), + expected_current_version_id: dst_opts.expected_current_version_id.clone(), ..Default::default() }; @@ -1087,9 +1090,12 @@ impl ECStore { return Ok(ObjectInfo::default()); } - let _object_lock_guard = self - .acquire_object_write_lock_if_needed("delete_object", bucket, object, &mut opts) - .await?; + let _object_lock_guard = if opts.expected_current_version_id.is_none() { + self.acquire_object_write_lock_if_needed("delete_object", bucket, object, &mut opts) + .await? + } else { + None + }; if opts.delete_prefix { self.delete_prefix(bucket, object, &opts).await?; diff --git a/docs/operations/atomic-object-undo.md b/docs/operations/atomic-object-undo.md new file mode 100644 index 000000000..94944ec21 --- /dev/null +++ b/docs/operations/atomic-object-undo.md @@ -0,0 +1,41 @@ +# Atomic object undo precondition + +RustFS supports a destination-side version precondition for the two S3 +operations used to undo changes in a versioned bucket: + +```text +x-rustfs-expected-current-version-id: +``` + +This is a RustFS extension, not a standard S3 header. + +## Restore a historical object version + +Send a same-object `CopyObject` request whose copy source includes the +historical `versionId`, and set the extension header to the version that was +current when the undo was planned. RustFS holds the destination namespace write +lock while it compares the current version and creates the restored version. + +The request fails without creating a version when the source is not a +historical version, the source and destination differ, or the destination +current version no longer matches. + +## Remove a delete marker + +Send a version-specific `DeleteObject` request for the delete marker and set the +extension header to the same version ID. RustFS removes it only when it is still +the current version and is a delete marker. The namespace write lock covers the +check and deletion. + +## Responses and retries + +- A stale, missing, non-current, or non-delete-marker version returns HTTP 412 + `PreconditionFailed` and does not mutate object history. +- An empty or malformed version header returns HTTP 400 `InvalidArgument`. +- A mismatched operation shape returns HTTP 400 `InvalidRequest`. +- A successful retry using an old expected version returns HTTP 412 because the + first successful request changed the current version. + +Object Lock retention, legal hold, authorization, replication, and ordinary S3 +behavior remain unchanged. The extension only adds a precondition; it does not +bypass existing validation. diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 97dc82d23..6c9fa43e2 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -21,6 +21,7 @@ use super::storage_api::test::contract::{ bucket::{BucketOperations, BucketOptions, MakeBucketOptions}, list::ListOperations as _, multipart::MultipartOperations as _, + namespace::NamespaceLocking as _, object::{ObjectIO as _, ObjectOperations as _}, }; use super::storage_api::test::ecfs::FS; @@ -40,6 +41,7 @@ use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATC use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT}; use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager}; use rustfs_utils::http::{SUFFIX_FORCE_DELETE, insert_header}; +use rustfs_utils::path::encode_dir_object; use s3s::{S3Request, dto::*}; use serial_test::serial; use std::{ @@ -63,6 +65,7 @@ const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOU const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED"; const ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED"; const ENV_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE"; +const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id"; fn init_tracing() { INIT.call_once(|| {}); @@ -546,6 +549,227 @@ async fn copy_object_if_none_match_existing_destination_returns_precondition_fai ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"] +async fn undo_copy_requires_the_observed_current_version() { + let (_disk_paths, ecstore) = setup_test_env().await; + let fs = FS::new(); + let usecase = DefaultObjectUsecase::from_global(); + let bucket = format!("test-undo-copy-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object = "test/object.txt"; + create_test_bucket(&ecstore, bucket.as_str()).await; + + let put = |payload: &'static [u8]| { + PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(streaming_blob_from_bytes(payload))) + .content_length(Some(payload.len() as i64)) + .build() + .unwrap() + }; + let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT))) + .await + .expect("initial version should be written") + .output + .version_id + .expect("versioned PUT should return a version ID"); + let observed_current = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"replacement"), Method::PUT))) + .await + .expect("replacement version should be written") + .output + .version_id + .expect("versioned PUT should return a version ID"); + + let copy = || { + CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: bucket.clone().into(), + key: object.to_string().into(), + version_id: Some(original.clone().into()), + }) + .bucket(bucket.clone()) + .key(object.to_string()) + .build() + .unwrap() + }; + let mut stale_req = build_request(copy(), Method::PUT); + stale_req.headers.insert( + RUSTFS_EXPECTED_CURRENT_VERSION_ID, + HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(), + ); + let versions_before = live_object_version_count(&ecstore, bucket.as_str(), object).await; + let err = Box::pin(usecase.execute_copy_object(stale_req)) + .await + .expect_err("stale destination must fail"); + assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed); + assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before); + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"replacement"); + + let mut matching_req = build_request(copy(), Method::PUT); + matching_req + .headers + .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed_current).unwrap()); + Box::pin(usecase.execute_copy_object(matching_req)) + .await + .expect("matching destination version should permit undo copy"); + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"original"); + assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), object).await, versions_before + 1); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"] +async fn undo_delete_only_removes_the_current_matching_delete_marker() { + let (_disk_paths, ecstore) = setup_test_env().await; + let fs = FS::new(); + let usecase = DefaultObjectUsecase::from_global(); + let bucket = format!("test-undo-delete-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object = "test/object.txt"; + create_test_bucket(&ecstore, bucket.as_str()).await; + + let put = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(streaming_blob_from_bytes(b"visible again"))) + .content_length(Some(13)) + .build() + .unwrap(); + Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT))) + .await + .expect("object version should be written"); + let marker = Box::pin( + usecase.execute_delete_object(build_request( + DeleteObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .build() + .unwrap(), + Method::DELETE, + )), + ) + .await + .expect("delete marker should be created") + .output + .version_id + .expect("versioned delete should return the marker version"); + + let delete_marker = || { + DeleteObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .version_id(Some(marker.clone())) + .build() + .unwrap() + }; + let mut stale_req = build_request(delete_marker(), Method::DELETE); + stale_req.headers.insert( + RUSTFS_EXPECTED_CURRENT_VERSION_ID, + HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(), + ); + let err = Box::pin(usecase.execute_delete_object(stale_req)) + .await + .expect_err("mismatched delete marker must fail"); + assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed); + + let mut matching_req = build_request(delete_marker(), Method::DELETE); + matching_req + .headers + .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&marker).unwrap()); + Box::pin(usecase.execute_delete_object(matching_req)) + .await + .expect("current matching delete marker should be removed"); + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"visible again"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[serial] +#[ignore = "global-state object-version integration test: runs serialized in the CI ILM Integration (serial) lane"] +async fn undo_copy_serializes_a_concurrent_put_until_after_commit() { + let (_disk_paths, ecstore) = setup_test_env().await; + let bucket = format!("test-undo-race-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object = "test/object.txt"; + create_test_bucket(&ecstore, bucket.as_str()).await; + + let put = |payload: &'static [u8]| { + PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(streaming_blob_from_bytes(payload))) + .content_length(Some(payload.len() as i64)) + .build() + .unwrap() + }; + let fs = FS::new(); + let usecase = DefaultObjectUsecase::from_global(); + let original = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"original"), Method::PUT))) + .await + .expect("original should be written") + .output + .version_id + .expect("original should be versioned"); + let observed = Box::pin(usecase.execute_put_object(&fs, build_request(put(b"observed"), Method::PUT))) + .await + .expect("observed should be written") + .output + .version_id + .expect("observed should be versioned"); + let encoded_object = encode_dir_object(object); + let lock = ecstore + .new_ns_lock(bucket.as_str(), &encoded_object) + .await + .expect("namespace lock should be available"); + let blocker = lock + .get_write_lock(Duration::from_secs(10)) + .await + .expect("test should acquire blocker"); + + let mut undo_req = build_request( + CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: bucket.clone().into(), + key: object.to_string().into(), + version_id: Some(original.into()), + }) + .bucket(bucket.clone()) + .key(object.to_string()) + .build() + .unwrap(), + Method::PUT, + ); + undo_req + .headers + .insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(&observed).unwrap()); + + let undo = tokio::spawn(async move { + let usecase = DefaultObjectUsecase::from_global(); + Box::pin(usecase.execute_copy_object(undo_req)).await + }); + tokio::task::yield_now().await; + let put = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(streaming_blob_from_bytes(b"concurrent"))) + .content_length(Some(10)) + .build() + .unwrap(); + let concurrent_put = tokio::spawn(async move { + let fs = FS::new(); + let usecase = DefaultObjectUsecase::from_global(); + Box::pin(usecase.execute_put_object(&fs, build_request(put, Method::PUT))).await + }); + + drop(blocker); + let undo_result = undo.await.expect("undo task should not panic"); + let put_result = concurrent_put.await.expect("PUT task should not panic"); + assert!(put_result.is_ok(), "ordinary PUT should eventually succeed: {put_result:?}"); + if let Err(err) = undo_result { + assert_eq!(err.code(), &s3s::S3ErrorCode::PreconditionFailed); + } + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, b"concurrent"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[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#1148 (ilm-1)"] diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 05e753533..e00c75fe7 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -167,6 +167,7 @@ use s3s::stream::{ByteStream, DynByteStream, RemainingLength}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024; +const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id"; const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES"; const DEFAULT_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024; const PUT_EAGER_STATUS_ELIGIBLE: &str = "eligible"; @@ -2571,6 +2572,31 @@ fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions { opts } +fn expected_current_version_id(headers: &HeaderMap) -> S3Result> { + headers + .get(RUSTFS_EXPECTED_CURRENT_VERSION_ID) + .map(|value| { + let value = value + .to_str() + .map(str::trim) + .map_err(|_| s3_error!(InvalidArgument, "Invalid expected current version ID header"))?; + if value.eq_ignore_ascii_case("null") { + return Ok(Uuid::nil().to_string()); + } + Uuid::parse_str(value) + .map(|version| version.to_string()) + .map_err(|_| s3_error!(InvalidArgument, "Invalid expected current version ID header")) + }) + .transpose() +} + +fn validate_undo_delete_version(expected: Option<&str>, requested: Option<&str>) -> S3Result<()> { + if expected.is_some() && expected != requested { + return Err(s3_error!(PreconditionFailed)); + } + Ok(()) +} + fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError { match err { rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable { @@ -6023,12 +6049,21 @@ impl DefaultObjectUsecase { .map_err(ApiError::from)?; let cp_src_dst_same = path_join_buf(&[&src_bucket, &src_key]) == path_join_buf(&[&bucket, &key]); + let expected_current_version_id = expected_current_version_id(&req.headers)?; + if expected_current_version_id.is_some() + && (!cp_src_dst_same || version_id.is_none() || dest_version_id.is_some() || !dst_opts.versioned) + { + return Err(s3_error!( + InvalidRequest, + "Expected current version precondition requires a versioned same-object historical copy that creates a new version" + )); + } let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let _self_copy_lock_guard = if cp_src_dst_same { + let _self_copy_lock_guard = if cp_src_dst_same && expected_current_version_id.is_none() { let guard = acquire_self_copy_namespace_lock(store.as_ref(), &bucket, &key).await?; src_opts.no_lock = true; src_get_opts.no_lock = true; @@ -6037,21 +6072,30 @@ impl DefaultObjectUsecase { } else { None }; + dst_opts.expected_current_version_id = expected_current_version_id.clone(); let mut current_opts: ObjectOptions = internal_object_info_lookup_opts( get_opts(&bucket, &key, dest_version_id.clone(), None, &req.headers) .await .map_err(ApiError::from)?, ); - if cp_src_dst_same { + if _self_copy_lock_guard.is_some() { current_opts.no_lock = true; } let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { validate_existing_object_lock_for_write(&existing_obj_info, &dst_opts)?; + if let Some(expected) = expected_current_version_id.as_deref() + && existing_obj_info.version_id.unwrap_or_default().to_string() != expected + { + return Err(s3_error!(PreconditionFailed)); + } Some(existing_obj_info.size.max(0) as u64) } Err(err) => { + if expected_current_version_id.is_some() { + return Err(s3_error!(PreconditionFailed)); + } if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { return Err(ApiError::from(err).into()); } @@ -6898,9 +6942,18 @@ impl DefaultObjectUsecase { // } } + let expected_current_version_id = expected_current_version_id(&req.headers)?; + if expected_current_version_id.is_some() && (force_delete || !opts.versioned) { + return Err(s3_error!( + InvalidRequest, + "Expected current version precondition requires a version-specific delete in a versioned bucket" + )); + } + validate_undo_delete_version(expected_current_version_id.as_deref(), opts.version_id.as_deref())?; let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; + opts.expected_current_version_id = expected_current_version_id.clone(); let replicate_force_delete = force_delete && !replica @@ -6920,7 +6973,6 @@ impl DefaultObjectUsecase { let get_opts: ObjectOptions = get_opts(&bucket, &key, version_id_clone, None, &req.headers) .await .map_err(ApiError::from)?; - let existing_object_info = match store.get_object_info(&bucket, &key, &get_opts).await { Ok(obj_info) => { // Check for bypass governance retention header (permission already verified in access.rs) @@ -12930,6 +12982,70 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } + #[test] + fn expected_current_version_header_normalizes_uuid_and_null() { + let version = Uuid::new_v4(); + let mut headers = HeaderMap::new(); + headers.insert( + RUSTFS_EXPECTED_CURRENT_VERSION_ID, + HeaderValue::from_str(&version.to_string().to_uppercase()).unwrap(), + ); + assert_eq!(expected_current_version_id(&headers).unwrap(), Some(version.to_string())); + + headers.insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_static(" null ")); + assert_eq!(expected_current_version_id(&headers).unwrap(), Some(Uuid::nil().to_string())); + } + + #[test] + fn expected_current_version_header_rejects_empty_and_malformed_values() { + for value in ["", "not-a-version"] { + let mut headers = HeaderMap::new(); + headers.insert(RUSTFS_EXPECTED_CURRENT_VERSION_ID, HeaderValue::from_str(value).unwrap()); + assert_eq!(expected_current_version_id(&headers).unwrap_err().code(), &S3ErrorCode::InvalidArgument); + } + } + + #[tokio::test] + async fn execute_copy_object_rejects_expected_version_for_different_destination() { + let input = CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: "test-bucket".into(), + key: "source-key".into(), + version_id: Some(Uuid::new_v4().to_string().into()), + }) + .bucket("test-bucket".to_string()) + .key("destination-key".to_string()) + .build() + .unwrap(); + let mut req = build_request(input, Method::PUT); + req.headers.insert( + RUSTFS_EXPECTED_CURRENT_VERSION_ID, + HeaderValue::from_str(&Uuid::new_v4().to_string()).unwrap(), + ); + + let err = Box::pin(DefaultObjectUsecase::without_context().execute_copy_object(req)) + .await + .unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn undo_delete_requires_version_id_to_match_expected_current() { + let expected = Uuid::new_v4().to_string(); + assert!(validate_undo_delete_version(Some(&expected), Some(&expected)).is_ok()); + assert_eq!( + validate_undo_delete_version(Some(&expected), Some(&Uuid::new_v4().to_string())) + .unwrap_err() + .code(), + &S3ErrorCode::PreconditionFailed + ); + assert_eq!( + validate_undo_delete_version(Some(&expected), None).unwrap_err().code(), + &S3ErrorCode::PreconditionFailed + ); + assert!(validate_undo_delete_version(None, None).is_ok()); + } + #[tokio::test] async fn execute_delete_objects_rejects_empty_object_list() { let input = DeleteObjectsInput::builder() diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index f24fef4f4..88f1b491c 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1065,6 +1065,10 @@ pub(crate) mod test { pub(crate) use super::super::super::storage_contracts::MultipartOperations; } + pub(crate) mod namespace { + pub(crate) use super::super::super::storage_contracts::NamespaceLocking; + } + pub(crate) mod object { pub(crate) use super::super::super::storage_contracts::{ObjectIO, ObjectOperations}; }