From 8cb9296f204d7270255350118e582f55c482a132 Mon Sep 17 00:00:00 2001 From: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:33:46 +0530 Subject: [PATCH] perf(ecstore): reuse bucket incarnation fences across copy and metadata paths (#9034) Eliminate redundant .bucket-incarnation/xl.meta quorum reads and distributed bucket-metadata transaction read locks under held lifecycle read locks and Object Lock configuration snapshots across put_object_metadata, CopyObject, UploadPartCopy, RestoreObject, read_object_versions_for_lifecycle, and get_bucket_incarnation_id_for_options_in. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a --- .../src/bucket/lifecycle/metadata_boundary.rs | 9 +- crates/ecstore/src/bucket/metadata_sys.rs | 29 +++- crates/ecstore/src/set_disk/ops/object.rs | 13 +- crates/ecstore/src/store/bucket.rs | 76 ++++++++- crates/ecstore/src/store/bucket_fence.rs | 23 ++- crates/ecstore/src/store/init.rs | 150 ++++++++++++++++++ crates/ecstore/src/store/list_objects.rs | 15 +- crates/ecstore/src/store/multipart.rs | 2 +- crates/ecstore/src/store/object.rs | 69 +++----- rustfs/src/app/multipart_usecase.rs | 35 ++-- rustfs/src/app/object/copy.rs | 79 ++------- rustfs/src/app/object/mod.rs | 7 +- rustfs/src/app/object/restore.rs | 28 ++-- rustfs/src/app/storage_api.rs | 1 - 14 files changed, 359 insertions(+), 177 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs b/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs index 11363659d..c0c512e29 100644 --- a/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs +++ b/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs @@ -63,14 +63,7 @@ pub(crate) async fn get_expiry_configs_for_options( bucket: &str, opts: &crate::object_api::ObjectOptions, ) -> Result { - let bucket_incarnation_id = match opts - .object_lock_config_snapshot - .as_ref() - .and_then(|snapshot| snapshot.validated_bucket_incarnation_for(api.id, bucket, opts.expected_bucket_incarnation_id)) - { - Some(incarnation_id) => incarnation_id, - None => metadata_sys::get_bucket_incarnation_id_for_options_in(&api.ctx, bucket, opts).await?, - }; + let bucket_incarnation_id = metadata_sys::get_bucket_incarnation_id_for_options_in(&api.ctx, bucket, opts).await?; let metadata = get_authoritative_metadata(api, bucket, bucket_incarnation_id).await?; let table_bucket_enabled = metadata.table_bucket_enabled(); diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index cb2ad0fb3..99825a832 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -564,12 +564,39 @@ pub(crate) async fn validate_bucket_reconstruction_in( Ok(()) } -/// Validate against disk while retaining an already-held Object Lock fence. +/// Resolve the authoritative bucket incarnation id for `opts`: +/// 1. When namespace locking is enabled and `opts.object_lock_config_snapshot` is +/// present, validate and return the snapshot's incarnation under its held +/// lifecycle and metadata-transaction fences. +/// 2. When namespace locking is enabled and `opts.bucket_lifecycle_lock_fence` is +/// present and not lost, reuse the store's `BucketFenceRegistry` memo if a +/// registered lifecycle read guard is currently held without lock loss. +/// 3. Otherwise, acquire the bucket metadata transaction read lock (unless +/// already held on `opts`) and read the incarnation from disk. pub(crate) async fn get_bucket_incarnation_id_for_options_in( ctx: &crate::runtime::instance::InstanceContext, bucket: &str, opts: &ObjectOptions, ) -> Result { + if !ctx.lock_manager().is_disabled() { + if let Some(snapshot) = opts.object_lock_config_snapshot.as_ref() { + let store = object_store_in(ctx).await?; + return snapshot + .validated_bucket_incarnation_for(store.id, bucket, opts.expected_bucket_incarnation_id) + .ok_or_else(|| { + Error::other("Object Lock snapshot does not hold a valid metadata transaction fence for this bucket") + }); + } + if opts + .bucket_lifecycle_lock_fence + .as_ref() + .is_some_and(|fence| !fence.is_lock_lost()) + && let Some(store) = object_store_if_initialized_in(ctx).await + && let Some(incarnation) = store.bucket_fence_registry.validated_incarnation(bucket) + { + return Ok(incarnation); + } + } let sys = bucket_metadata_sys_of(ctx)?; let sys = sys.read().await.clone(); let guard = acquire_bucket_metadata_transaction_read_lock_for_options_in(ctx, bucket, opts).await?; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 6cb5e0b9b..bd860ded6 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -11958,13 +11958,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let bucket_lifecycle_guard = if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id && opts.bucket_lifecycle_lock_fence.is_none() { - let guard = metadata_sys::object_store_in(&self.ctx) - .await? - .acquire_bucket_lifecycle_read_lock(bucket) - .await?; - self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts) - .await?; - Some(guard) + Some( + metadata_sys::object_store_in(&self.ctx) + .await? + .acquire_bucket_incarnation_fence_with_options(bucket, expected_incarnation_id, Some(opts)) + .await?, + ) } else { None }; diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index b5a77b548..425a2671e 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -564,13 +564,83 @@ impl ECStore { .await } - pub(super) async fn acquire_bucket_incarnation_fence_with_options( + pub(crate) async fn acquire_bucket_incarnation_fence_with_options( &self, bucket: &str, expected: Uuid, opts: Option<&ObjectOptions>, ) -> Result { + Ok(self + .acquire_bucket_incarnation_fence_for_options(bucket, Some(expected), opts) + .await? + .0) + } + + pub(crate) async fn acquire_bucket_incarnation_fence_for_options( + &self, + bucket: &str, + expected: Option, + opts: Option<&ObjectOptions>, + ) -> Result<(super::BucketIncarnationFenceGuard, Uuid)> { let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?; + self.validate_bucket_incarnation_fence_guard_with_options(bucket, inner, expected, opts) + .await + } + + /// Acquire lexicographically ordered bucket lifecycle read locks for a copy + /// operation and validate both bucket incarnations through [`super::BucketFenceRegistry`]. + pub async fn acquire_copy_bucket_incarnation_fences( + &self, + source_bucket: &str, + expected_source: Option, + destination_bucket: &str, + expected_destination: Option, + ) -> Result<(super::BucketIncarnationFenceGuard, Uuid, Option, Uuid)> { + let (source_guard, destination_guard) = if source_bucket == destination_bucket { + (self.acquire_bucket_lifecycle_read_lock(source_bucket).await?, None) + } else if source_bucket < destination_bucket { + let source_guard = self.acquire_bucket_lifecycle_read_lock(source_bucket).await?; + let destination_guard = self.acquire_bucket_lifecycle_read_lock(destination_bucket).await?; + (source_guard, Some(destination_guard)) + } else { + let destination_guard = self.acquire_bucket_lifecycle_read_lock(destination_bucket).await?; + let source_guard = self.acquire_bucket_lifecycle_read_lock(source_bucket).await?; + (source_guard, Some(destination_guard)) + }; + let (source_guard, current_source_incarnation_id) = self + .validate_bucket_incarnation_fence_guard_with_options(source_bucket, source_guard, expected_source, None) + .await?; + let (destination_guard, current_destination_incarnation_id) = if let Some(destination_guard) = destination_guard { + let (destination_guard, destination_incarnation_id) = self + .validate_bucket_incarnation_fence_guard_with_options( + destination_bucket, + destination_guard, + expected_destination, + None, + ) + .await?; + (Some(destination_guard), destination_incarnation_id) + } else { + if expected_destination.is_some_and(|expected| expected != current_source_incarnation_id) { + return Err(StorageError::BucketNotFound(destination_bucket.to_owned())); + } + (None, current_source_incarnation_id) + }; + Ok(( + source_guard, + current_source_incarnation_id, + destination_guard, + current_destination_incarnation_id, + )) + } + + async fn validate_bucket_incarnation_fence_guard_with_options( + &self, + bucket: &str, + inner: rustfs_lock::NamespaceLockGuard, + expected: Option, + opts: Option<&ObjectOptions>, + ) -> Result<(super::BucketIncarnationFenceGuard, Uuid)> { let mut guard = super::BucketIncarnationFenceGuard::new( Arc::clone(&self.bucket_fence_registry), bucket, @@ -586,10 +656,10 @@ impl ECStore { } }) .await?; - if current != expected { + if expected.is_some_and(|expected| current != expected) { return Err(StorageError::BucketNotFound(bucket.to_owned())); } - Ok(guard) + Ok((guard, current)) } /// Hold this guard through recursive-delete authorization and mutation so diff --git a/crates/ecstore/src/store/bucket_fence.rs b/crates/ecstore/src/store/bucket_fence.rs index 9c51d8290..d5238dc05 100644 --- a/crates/ecstore/src/store/bucket_fence.rs +++ b/crates/ecstore/src/store/bucket_fence.rs @@ -149,6 +149,18 @@ impl BucketFenceRegistry { } } } + + /// Return the memoized incarnation id for `bucket` when at least one + /// registered lifecycle read guard is currently held without lock loss. + pub(crate) fn validated_incarnation(&self, bucket: &str) -> Option { + let mut entries = self.entries.lock().expect("bucket fence registry poisoned"); + let entry = entries.get_mut(bucket)?; + if entry.guards.iter().any(|guard| guard.loss_probe.is_lost()) { + entry.validated = None; + return None; + } + entry.validated + } } /// Owns continuous lifecycle coverage and its incarnation validation memo. @@ -221,7 +233,7 @@ impl BucketIncarnationFenceGuard { } } - pub(crate) fn is_lock_lost(&self) -> bool { + pub fn is_lock_lost(&self) -> bool { self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) } @@ -310,9 +322,16 @@ mod tests { let (first_probe, first_lost) = controllable_probe(); let first = reg.enter("b", first_probe); assert_eq!(first.memoized, None); + assert_eq!(reg.validated_incarnation("b"), None); reg.memoize("b", uuid(7)); + assert_eq!(reg.validated_incarnation("b"), Some(uuid(7))); first_lost.store(true, std::sync::atomic::Ordering::SeqCst); + assert_eq!( + reg.validated_incarnation("b"), + None, + "live lost guard must clear and block validated_incarnation" + ); let second = reg.enter("b", live_probe()); assert_eq!(second.memoized, None, "live lost guard must force disk revalidation"); @@ -320,9 +339,11 @@ mod tests { let third = reg.enter("b", live_probe()); assert_eq!(third.memoized, None, "memo remains blocked while the lost guard is live"); + assert_eq!(reg.validated_incarnation("b"), None); reg.exit("b", first.token, true); reg.memoize("b", uuid(8)); + assert_eq!(reg.validated_incarnation("b"), Some(uuid(8))); let fourth = reg.enter("b", live_probe()); assert_eq!(fourth.memoized, Some(uuid(8)), "memo resumes after lost coverage leaves"); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a4a392f5b..dde39f37d 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -21384,6 +21384,141 @@ mod tests { shutdown.cancel(); } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn put_object_metadata_and_copy_object_reuse_overlapping_incarnation_fence() { + let temp = tempfile::tempdir().expect("create metadata/copy fence store"); + let (_ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp.path(), "metadata-copy-fence-reuse", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("metadata-copy-fence-{}", Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket"); + let mut seed_reader = PutObjReader::from_vec(b"seed payload".to_vec()); + let versioned_opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + let mut src_info = store + .put_object(&bucket, "src.bin", &mut seed_reader, &versioned_opts) + .await + .expect("seed source object"); + let src_version_id = src_info + .version_id + .expect("versioned PUT should return a version ID") + .to_string(); + let expected = store.bucket_incarnation_id(&bucket).await.expect("read incarnation"); + let coverage = store + .acquire_bucket_incarnation_fence(&bucket, expected) + .await + .expect("validate continuous coverage"); + let metadata_writer = crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_lock_in(&store.ctx, &bucket) + .await + .expect("hold metadata writer during covered metadata and copy operations"); + + let updated = tokio::time::timeout( + Duration::from_secs(10), + store.put_object_metadata( + &bucket, + "src.bin", + &ObjectOptions { + version_id: Some(src_version_id.clone()), + versioned: true, + eval_metadata: Some(HashMap::from([("x-amz-meta-fenced".to_string(), "true".to_string())])), + ..Default::default() + }, + ), + ) + .await + .expect("covered put_object_metadata must not reacquire the metadata transaction lock") + .expect("update object metadata using authoritative memo"); + assert_eq!(updated.user_defined.get("x-amz-meta-fenced").map(String::as_str), Some("true")); + + Arc::make_mut(&mut src_info.user_defined).insert("x-amz-meta-copied".to_string(), "true".to_string()); + let src_copy_opts = ObjectOptions { + version_id: Some(src_version_id), + versioned: true, + ..Default::default() + }; + let copied = tokio::time::timeout( + Duration::from_secs(10), + store.copy_object(&bucket, "src.bin", &bucket, "src.bin", &mut src_info, &src_copy_opts, &versioned_opts), + ) + .await + .expect("covered versioned copy_object must not reacquire the metadata transaction lock") + .expect("versioned copy using authoritative memo"); + assert_eq!(copied.user_defined.get("x-amz-meta-copied").map(String::as_str), Some("true")); + + let (src_guard, src_incarnation, dst_guard, dst_incarnation) = tokio::time::timeout( + Duration::from_secs(10), + store.acquire_copy_bucket_incarnation_fences(&bucket, Some(expected), &bucket, Some(expected)), + ) + .await + .expect("covered acquire_copy_bucket_incarnation_fences must not reacquire the metadata transaction lock") + .expect("validate same-bucket copy fences using authoritative memo"); + assert!(dst_guard.is_none()); + assert_eq!(src_incarnation, expected); + assert_eq!(dst_incarnation, expected); + let mut borrowed_fence_opts = ObjectOptions { + expected_bucket_incarnation_id: Some(expected), + ..Default::default() + }; + src_guard.attach_to_object_options(&mut borrowed_fence_opts); + let fenced_incarnation = tokio::time::timeout( + Duration::from_secs(10), + crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&store.ctx, &bucket, &borrowed_fence_opts), + ) + .await + .expect("borrowed fence under active coverage must not reacquire the metadata transaction lock") + .expect("resolve incarnation under borrowed fence"); + assert_eq!(fenced_incarnation, expected); + assert!( + matches!( + store + .acquire_copy_bucket_incarnation_fences(&bucket, Some(expected), &bucket, Some(Uuid::new_v4())) + .await, + Err(StorageError::BucketNotFound(_)) + ), + "same-bucket copy fence must reject a mismatched destination incarnation" + ); + drop(src_guard); + + let stale_opts = ObjectOptions { + expected_bucket_incarnation_id: Some(Uuid::new_v4()), + versioned: true, + ..Default::default() + }; + assert!( + matches!( + store.put_object_metadata(&bucket, "src.bin", &stale_opts).await, + Err(StorageError::BucketNotFound(_)) + ), + "put_object_metadata must reject a stale expected incarnation even under warm coverage" + ); + assert!( + matches!( + store + .copy_object( + &bucket, + "src.bin", + &bucket, + "src.bin", + &mut src_info, + &ObjectOptions::default(), + &stale_opts, + ) + .await, + Err(StorageError::BucketNotFound(_)) + ), + "copy_object must reject a stale expected incarnation even under warm coverage" + ); + drop(metadata_writer); + drop(coverage); + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn multipart_snapshot_bypasses_initializer_queued_behind_metadata_writer() { @@ -21469,10 +21604,25 @@ mod tests { .acquire_bucket_incarnation_fence(&bucket, old) .await .expect("validate with no-op lock"); + let snapshot = store + .object_lock_config_snapshot_under_lifecycle_fence(&bucket, &crate::object_api::NamespaceLockFence::new()) + .await + .expect("capture snapshot under no-op lock"); let current = Uuid::new_v4(); crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, current) .await .expect("replace authority while no-op guard provides no exclusion"); + let snapshot_opts = ObjectOptions { + expected_bucket_incarnation_id: Some(old), + object_lock_config_snapshot: Some(snapshot), + ..Default::default() + }; + assert_eq!( + crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&store.ctx, &bucket, &snapshot_opts) + .await + .expect("disabled lock manager must re-read incarnation from disk instead of trusting snapshot"), + current + ); assert!( matches!( store.acquire_bucket_incarnation_fence(&bucket, old).await, diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 1664a551d..f9aad4717 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -1030,16 +1030,19 @@ impl ECStore { // Lock order matches writes: bucket lifecycle, metadata validation, // then object namespace. Borrow fences supplied by an outer operation. let bucket_guard = if opts.bucket_lifecycle_lock_fence.is_none() { - let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?; - opts.add_bucket_lifecycle_lock_guard(&guard); + let guard = self + .acquire_bucket_incarnation_fence_with_options(bucket, bucket_incarnation_id, Some(&opts)) + .await?; + guard.attach_to_object_options(&mut opts); Some(guard) } else { + let incarnation = + crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?; + if incarnation != bucket_incarnation_id { + return Err(Error::BucketNotFound(bucket.to_owned())); + } None }; - let incarnation = crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?; - if incarnation != bucket_incarnation_id { - return Err(Error::BucketNotFound(bucket.to_owned())); - } let metadata_sys = crate::bucket::metadata_sys::bucket_metadata_sys_of(&self.ctx)?; let metadata_sys = metadata_sys.read().await.clone(); let (versioning, _) = metadata_sys.get_versioning_config(bucket).await?; diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 2b96a41f5..3646ae5cd 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -412,7 +412,7 @@ impl ECStore { /// Multipart lock order is bucket lifecycle, generation validation, then /// object/upload locks in the selected set. - async fn guard_multipart_bucket_incarnation( + pub(super) async fn guard_multipart_bucket_incarnation( &self, bucket: &str, opts: &ObjectOptions, diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 7949fe82b..2e54400cb 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -30,8 +30,8 @@ use crate::bucket::lifecycle::{ }, }; use crate::bucket::metadata_sys::{ - acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_for_options_in, - get_cached_bucket_incarnation_id_in, get_delete_config_from_disk_in, get_object_lock_config_and_incarnation_from_disk_in, + acquire_bucket_metadata_transaction_read_lock_in, get_cached_bucket_incarnation_id_in, get_delete_config_from_disk_in, + get_object_lock_config_and_incarnation_from_disk_in, }; use crate::bucket::object_lock::objectlock_sys::{ check_object_lock_for_deletion_with_state, ensure_recursive_force_delete_allowed_for_state, @@ -4883,24 +4883,20 @@ impl ECStore { if !is_meta_bucketname(dst_bucket) && dst_opts.expected_bucket_incarnation_id.is_none() { dst_opts.expected_bucket_incarnation_id = Some(self.bucket_incarnation_id(dst_bucket).await?); } - let _bucket_lifecycle_guard = if is_meta_bucketname(dst_bucket) || dst_opts.bucket_lifecycle_lock_fence.is_some() { - None - } else { - Some(self.acquire_bucket_lifecycle_read_lock(dst_bucket).await?) - }; - let current_bucket_incarnation_id = if let Some(guard) = _bucket_lifecycle_guard.as_ref() { - dst_opts.add_bucket_lifecycle_lock_guard(guard); - let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, dst_bucket, &dst_opts).await?; - if dst_opts - .expected_bucket_incarnation_id - .is_some_and(|expected| expected != current_incarnation_id) - { - return Err(StorageError::BucketNotFound(dst_bucket.to_string())); - } - Some(current_incarnation_id) - } else { - dst_opts.expected_bucket_incarnation_id - }; + let (_bucket_lifecycle_guard, current_bucket_incarnation_id) = + if is_meta_bucketname(dst_bucket) || dst_opts.bucket_lifecycle_lock_fence.is_some() { + (None, dst_opts.expected_bucket_incarnation_id) + } else { + let (guard, current_incarnation_id) = self + .acquire_bucket_incarnation_fence_for_options( + dst_bucket, + dst_opts.expected_bucket_incarnation_id, + Some(&dst_opts), + ) + .await?; + guard.attach_to_object_options(&mut dst_opts); + (Some(guard), Some(current_incarnation_id)) + }; if dst_opts .bucket_lifecycle_lock_fence .as_ref() @@ -6504,24 +6500,8 @@ impl ECStore { opts: &ObjectOptions, ) -> Result<()> { let object = encode_dir_object(object); - let mut opts = transition_restore_pool_opts(opts); - if !is_meta_bucketname(bucket) && opts.expected_bucket_incarnation_id.is_none() { - opts.expected_bucket_incarnation_id = Some(self.bucket_incarnation_id(bucket).await?); - } - let bucket_lifecycle_guard = if is_meta_bucketname(bucket) { - None - } else { - Some(self.acquire_bucket_lifecycle_read_lock(bucket).await?) - }; - if let Some(guard) = bucket_lifecycle_guard.as_ref() { - opts.add_bucket_lifecycle_lock_guard(guard); - } - if !is_meta_bucketname(bucket) { - let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?; - if opts.expected_bucket_incarnation_id != Some(current_incarnation_id) { - return Err(StorageError::BucketNotFound(bucket.to_string())); - } - } + let opts = transition_restore_pool_opts(opts); + let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, &opts).await?; if opts.overwrites_existing_version() && !is_meta_bucketname(bucket) { let expected_incarnation_id = opts .expected_bucket_incarnation_id @@ -6589,16 +6569,11 @@ impl ECStore { let bucket_lifecycle_guard = if is_meta_bucketname(bucket) { None } else { - let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?; - let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?; - if opts - .expected_bucket_incarnation_id - .is_some_and(|expected| expected != current_incarnation_id) - { - return Err(StorageError::BucketNotFound(bucket.to_string())); - } + let (guard, current_incarnation_id) = self + .acquire_bucket_incarnation_fence_for_options(bucket, opts.expected_bucket_incarnation_id, Some(&opts)) + .await?; opts.expected_bucket_incarnation_id = Some(current_incarnation_id); - opts.add_bucket_lifecycle_lock_guard(&guard); + guard.attach_to_object_options(&mut opts); if guard.is_lock_lost() { return Err(Error::other("bucket lifecycle lock was lost before the metadata update")); } diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 6f555700e..d7f849eb0 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -79,8 +79,8 @@ use crate::app::object_data_cache::{ invalidate_object_data_cache_before_mutation, }; use crate::app::object_usecase::{ - acquire_copy_bucket_lifecycle_locks, apply_quota_admission, build_put_like_object_lock_metadata, map_quota_check_outcome, - s3s_body_error_to_io, validate_existing_object_lock_for_write, + apply_quota_admission, build_put_like_object_lock_metadata, map_quota_check_outcome, s3s_body_error_to_io, + validate_existing_object_lock_for_write, }; use crate::app::runtime_sources::{ AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context, @@ -1727,23 +1727,20 @@ impl DefaultMultipartUsecase { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let (source_bucket_lifecycle_guard, destination_bucket_lifecycle_guard_storage) = - acquire_copy_bucket_lifecycle_locks(store.as_ref(), &src_bucket, &bucket).await?; - let current_source_incarnation_id = store - .bucket_incarnation_id_from_disk(&src_bucket) + let ( + source_bucket_lifecycle_guard, + current_source_incarnation_id, + destination_bucket_lifecycle_guard_storage, + current_destination_incarnation_id, + ) = store + .acquire_copy_bucket_incarnation_fences( + &src_bucket, + expected_source_incarnation_id, + &bucket, + expected_destination_incarnation_id, + ) .await .map_err(ApiError::from)?; - if expected_source_incarnation_id.is_some_and(|expected| expected != current_source_incarnation_id) { - return Err(ApiError::from(StorageError::BucketNotFound(src_bucket.clone())).into()); - } - let current_destination_incarnation_id = if src_bucket == bucket { - current_source_incarnation_id - } else { - store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)? - }; - if expected_destination_incarnation_id.is_some_and(|expected| expected != current_destination_incarnation_id) { - return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into()); - } let destination_bucket_lifecycle_guard = destination_bucket_lifecycle_guard_storage .as_ref() .unwrap_or(&source_bucket_lifecycle_guard); @@ -1762,9 +1759,9 @@ impl DefaultMultipartUsecase { ..Default::default() }; if src_bucket != bucket { - dst_opts.add_bucket_lifecycle_lock_guard(&source_bucket_lifecycle_guard); + source_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts); } - dst_opts.add_bucket_lifecycle_lock_guard(destination_bucket_lifecycle_guard); + destination_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts); let mp_info = store .get_multipart_info(&bucket, &key, &upload_id, &dst_opts) diff --git a/rustfs/src/app/object/copy.rs b/rustfs/src/app/object/copy.rs index 42158aa65..1e6b3ce43 100644 --- a/rustfs/src/app/object/copy.rs +++ b/rustfs/src/app/object/copy.rs @@ -124,48 +124,6 @@ where .map_err(|err| ApiError::from(copy_namespace_lock_error(bucket, &object, "write", err)).into()) } -pub(crate) async fn acquire_copy_bucket_lifecycle_lock(store: &S, bucket: &str) -> S3Result -where - S: NamespaceLocking + ?Sized, -{ - let lock = store - .new_ns_lock(bucket, BUCKET_LIFECYCLE_LOCK_OBJECT) - .await - .map_err(ApiError::from)?; - lock.get_read_lock(get_lock_acquire_timeout()).await.map_err(|err| { - ApiError::from(copy_namespace_lock_error( - bucket, - BUCKET_LIFECYCLE_LOCK_OBJECT, - "bucket_lifecycle_read", - err, - )) - .into() - }) -} - -pub(crate) async fn acquire_copy_bucket_lifecycle_locks( - store: &S, - source_bucket: &str, - destination_bucket: &str, -) -> S3Result<(NamespaceLockGuard, Option)> -where - S: NamespaceLocking + ?Sized, -{ - if source_bucket == destination_bucket { - return Ok((acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?, None)); - } - - if source_bucket < destination_bucket { - let source_guard = acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?; - let destination_guard = acquire_copy_bucket_lifecycle_lock(store, destination_bucket).await?; - Ok((source_guard, Some(destination_guard))) - } else { - let destination_guard = acquire_copy_bucket_lifecycle_lock(store, destination_bucket).await?; - let source_guard = acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?; - Ok((source_guard, Some(destination_guard))) - } -} - impl DefaultObjectUsecase { pub fn execute_copy_object( &self, @@ -383,29 +341,20 @@ impl DefaultObjectUsecase { let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let (source_bucket_lifecycle_guard, destination_bucket_lifecycle_guard_storage) = - acquire_copy_bucket_lifecycle_locks(store.as_ref(), &src_bucket, &bucket).await?; - let current_source_incarnation_id = store - .bucket_incarnation_id_from_disk(&src_bucket) + let ( + source_bucket_lifecycle_guard, + current_source_incarnation_id, + destination_bucket_lifecycle_guard_storage, + current_destination_incarnation_id, + ) = store + .acquire_copy_bucket_incarnation_fences( + &src_bucket, + src_get_opts.expected_bucket_incarnation_id, + &bucket, + dst_opts.expected_bucket_incarnation_id, + ) .await .map_err(ApiError::from)?; - if src_get_opts - .expected_bucket_incarnation_id - .is_some_and(|expected| expected != current_source_incarnation_id) - { - return Err(ApiError::from(StorageError::BucketNotFound(src_bucket.clone())).into()); - } - let current_destination_incarnation_id = if src_bucket == bucket { - current_source_incarnation_id - } else { - store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)? - }; - if dst_opts - .expected_bucket_incarnation_id - .is_some_and(|expected| expected != current_destination_incarnation_id) - { - return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into()); - } let destination_bucket_lifecycle_guard = destination_bucket_lifecycle_guard_storage .as_ref() .unwrap_or(&source_bucket_lifecycle_guard); @@ -422,9 +371,9 @@ impl DefaultObjectUsecase { src_get_opts.expected_bucket_incarnation_id = Some(current_source_incarnation_id); dst_opts.expected_bucket_incarnation_id = Some(current_destination_incarnation_id); if src_bucket != bucket { - dst_opts.add_bucket_lifecycle_lock_guard(&source_bucket_lifecycle_guard); + source_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts); } - dst_opts.add_bucket_lifecycle_lock_guard(destination_bucket_lifecycle_guard); + destination_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts); // Bucket metadata uses the bucket name as its namespace-lock key. Load // every copy-time bucket snapshot before a same-object key can collide diff --git a/rustfs/src/app/object/mod.rs b/rustfs/src/app/object/mod.rs index 378aad107..03f6f3e3f 100644 --- a/rustfs/src/app/object/mod.rs +++ b/rustfs/src/app/object/mod.rs @@ -134,7 +134,6 @@ use rustfs_config::MI_B; use rustfs_filemeta::{NULL_VERSION_ID, RestoreStatusOps, parse_restore_obj_status}; use rustfs_io_core::{BytesPool, PooledBuffer}; use rustfs_io_metrics; -use rustfs_lock::NamespaceLockGuard; use rustfs_notify::EventArgsBuilder; use rustfs_object_capacity::capacity_manager::get_capacity_manager; use rustfs_policy::policy::action::{Action, S3Action}; @@ -221,7 +220,6 @@ pub(crate) mod shared; #[cfg(test)] mod test_support; -pub(crate) use self::copy::*; #[cfg(test)] pub(crate) use self::delete::*; pub(crate) use self::extract::*; @@ -264,9 +262,8 @@ use tracing::{debug, error, instrument, warn}; use uuid::Uuid; use super::storage_api::object_usecase::{ - BUCKET_LIFECYCLE_LOCK_OBJECT, GetObjectReader, StorageDeletedObject, StorageObjectInfo as ObjectInfo, - StorageObjectLockDeleteOptions, StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, - StoragePutObjReader as PutObjReader, + GetObjectReader, StorageDeletedObject, StorageObjectInfo as ObjectInfo, StorageObjectLockDeleteOptions, + StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader, }; use crate::app::object_data_cache::{ ColdFillCoordinateOutcome, ColdFillDiskPermitOwner, ColdFillError, ColdFillProducer, GetObjectBodyCacheLookup, diff --git a/rustfs/src/app/object/restore.rs b/rustfs/src/app/object/restore.rs index 412c1a1fd..53ea41411 100644 --- a/rustfs/src/app/object/restore.rs +++ b/rustfs/src/app/object/restore.rs @@ -213,8 +213,8 @@ impl DefaultObjectUsecase { // `apply_bucket_generation_guard` deliberately tolerates a missing guard // (only the S3 access layer installs one), so this must not hard-require // it. Resolve the current generation instead, exactly as the copy path - // does. The fence is unaffected: the value is re-read from disk and - // compared below, before the restore is admitted. + // does. The fence is unaffected: the value is validated under the + // lifecycle fence below, before the restore is admitted. let restore_bucket_incarnation_id = match opts.expected_bucket_incarnation_id { Some(incarnation_id) => incarnation_id, None => { @@ -246,10 +246,12 @@ 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 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 mut restore_bucket_lifecycle_guard = Some( + store + .acquire_bucket_incarnation_fence(&bucket, restore_bucket_incarnation_id) + .await + .map_err(ApiError::from)?, + ); let mut accept_guard = { let guard = store .acquire_restore_accept_guard(&bucket, &object) @@ -299,12 +301,12 @@ impl DefaultObjectUsecase { 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()); - } + restore_bucket_lifecycle_guard = Some( + store + .acquire_bucket_incarnation_fence(&bucket, restore_bucket_incarnation_id) + .await + .map_err(ApiError::from)?, + ); accept_guard = Some( store .acquire_restore_accept_guard(&bucket, &object) @@ -398,7 +400,7 @@ impl DefaultObjectUsecase { ..Default::default() }; if let Some(guard) = restore_bucket_lifecycle_guard.as_ref() { - restore_dst_opts.add_bucket_lifecycle_lock_guard(guard); + guard.attach_to_object_options(&mut restore_dst_opts); } if let Some(guard) = accept_guard.as_ref() { guard.add_namespace_lock_fence(&mut restore_dst_opts); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 3ac954919..21725a93c 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1203,7 +1203,6 @@ pub(crate) mod bucket_usecase { } pub(crate) mod object_usecase { - pub(crate) use super::storage_contracts::BUCKET_LIFECYCLE_LOCK_OBJECT; pub(crate) use super::storage_contracts::object::DeleteObjectCondition; pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};