Merge remote-tracking branch 'origin/main' into overtrue/fix-heal-lock-ci

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<LifecycleExpiryConfigs> {
-    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();
 

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<Uuid> {
+    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?;

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

@@ -11962,13 +11962,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
         };

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<super::BucketIncarnationFenceGuard> {
+        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<Uuid>,
+        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<Uuid>,
+        destination_bucket: &str,
+        expected_destination: Option<Uuid>,
+    ) -> Result<(super::BucketIncarnationFenceGuard, Uuid, Option<super::BucketIncarnationFenceGuard>, 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<Uuid>,
+        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

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<Uuid> {
+        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");
 

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,

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?;

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,

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"));
             }

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)

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<S>(store: &S, bucket: &str) -> S3Result<NamespaceLockGuard>
-where
-    S: NamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper> + ?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<S>(
-    store: &S,
-    source_bucket: &str,
-    destination_bucket: &str,
-) -> S3Result<(NamespaceLockGuard, Option<NamespaceLockGuard>)>
-where
-    S: NamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper> + ?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

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,

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);

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};