fix(ecstore): reuse metadata read guards during writes (#7941)

.config/nextest.toml

@@ -132,6 +132,12 @@ test-group = 'ecstore-serial-flaky'
 filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))'
 test-group = 'embedded-test-ports'
 
+# The real object probe includes bucket creation and cleanup in its short
+# measurement budget. Keep competing storage fixtures outside that budget.
+[[profile.default.overrides]]
+filter = 'package(rustfs) & binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)'
+threads-required = "num-test-threads"
+
 # Serialize the durable manual-transition checkpoint test across nextest's
 # process boundary; it mutates bucket lifecycle metadata and is not quarantined.
 [[profile.default.overrides]]
@@ -327,6 +333,10 @@ test-group = 'ecstore-serial-flaky'
 filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))'
 test-group = 'embedded-test-ports'
 
+[[profile.ci.overrides]]
+filter = 'package(rustfs) & binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)'
+threads-required = "num-test-threads"
+
 # Serialize the durable manual-transition checkpoint test under the ci profile
 # too. No retries: failures stay visible.
 [[profile.ci.overrides]]

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -3757,11 +3757,8 @@ pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) {
     }
 }
 
-pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) {
-    let Some(api) = runtime_sources::object_store_handle() else {
-        return;
-    };
-    let configs = match metadata_boundary::get_expiry_configs(&api, &oi.bucket).await {
+pub(crate) async fn enqueue_immediate_expiry(api: Arc<ECStore>, oi: &ObjectInfo, src: LcEventSrc, opts: &ObjectOptions) {
+    let configs = match metadata_boundary::get_expiry_configs_for_options(&api, &oi.bucket, opts).await {
         Ok(configs) => configs,
         Err(err) => {
             observe_lifecycle_observability_event(EVENT_LIFECYCLE_EVALUATION_FAILED, "failed", Some("metadata_unavailable"));
@@ -12703,7 +12700,8 @@ mod tests {
                 .push((event, state, reason));
         });
 
-        super::enqueue_immediate_expiry(&object_info, LcEventSrc::S3PutObject).await;
+        super::enqueue_immediate_expiry(Arc::clone(&ecstore), &object_info, LcEventSrc::S3PutObject, &ObjectOptions::default())
+            .await;
 
         assert!(
             observed.lock().expect("observed events should not poison").contains(&(

crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs

@@ -55,7 +55,15 @@ pub(crate) async fn lifecycle_expiry_allowed(
 }
 
 pub(crate) async fn get_expiry_configs(api: &crate::store::ECStore, bucket: &str) -> Result<LifecycleExpiryConfigs> {
-    let bucket_incarnation_id = api.bucket_incarnation_id_from_disk(bucket).await?;
+    get_expiry_configs_for_options(api, bucket, &crate::object_api::ObjectOptions::default()).await
+}
+
+pub(crate) async fn get_expiry_configs_for_options(
+    api: &crate::store::ECStore,
+    bucket: &str,
+    opts: &crate::object_api::ObjectOptions,
+) -> Result<LifecycleExpiryConfigs> {
+    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

@@ -24,6 +24,7 @@ use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_p
 use crate::bucket::utils::is_meta_bucketname;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result, is_err_bucket_not_found, is_err_strict_volume_not_found};
+use crate::object_api::ObjectOptions;
 use crate::runtime::sources as runtime_sources;
 use crate::storage_api_contracts::heal::HealOperations as _;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
@@ -490,6 +491,18 @@ pub(crate) async fn get_bucket_incarnation_id_in(ctx: &crate::runtime::instance:
     sys.get_bucket_incarnation_id_from_disk(bucket).await
 }
 
+/// Validate against disk while retaining an already-held Object Lock fence.
+pub(crate) async fn get_bucket_incarnation_id_for_options_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<Uuid> {
+    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?;
+    sys.get_bucket_incarnation_id_under_transaction_lock(bucket, &guard).await
+}
+
 pub(crate) async fn get_cached_bucket_incarnation_id_in(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
@@ -1055,6 +1068,24 @@ pub(crate) async fn acquire_bucket_metadata_transaction_read_lock_in(
     Ok(lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?)
 }
 
+/// Readers already holding an Object Lock snapshot must share its guard:
+/// a fresh read acquisition can queue behind a writer waiting for that snapshot.
+pub(crate) async fn acquire_bucket_metadata_transaction_read_lock_for_options_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    opts: &ObjectOptions,
+) -> Result<Arc<rustfs_lock::NamespaceLockGuard>> {
+    if let Some(snapshot) = opts.object_lock_config_snapshot.as_ref() {
+        let store = object_store_in(ctx).await?;
+        return snapshot
+            .metadata_transaction_guard_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")
+            });
+    }
+    Ok(Arc::new(acquire_bucket_metadata_transaction_read_lock_in(ctx, bucket).await?))
+}
+
 async fn acquire_transaction_lock_with_sys(
     sys: &Arc<RwLock<BucketMetadataSys>>,
     bucket: &str,
@@ -2357,8 +2388,17 @@ impl BucketMetadataSys {
         let _transaction_guard = transaction_lock
             .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
             .await?;
+        self.get_bucket_incarnation_id_under_transaction_lock(bucket, &_transaction_guard)
+            .await
+    }
+
+    async fn get_bucket_incarnation_id_under_transaction_lock(
+        &self,
+        bucket: &str,
+        transaction_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<Uuid> {
         let incarnation_id = load_bucket_incarnation(self.object_store(), bucket).await?;
-        if _transaction_guard.is_lock_lost() {
+        if transaction_guard.is_lock_lost() {
             return Err(Error::other(format!("bucket incarnation metadata transaction lock was lost: {bucket}")));
         }
         match incarnation_id {

crates/ecstore/src/bucket/quota/reservation.rs

@@ -47,6 +47,8 @@ static FAIL_NEXT_LEDGER_SAVE: std::sync::atomic::AtomicBool = std::sync::atomic:
 
 // Lock order: caller-held destination object/upload, bucket metadata
 // transaction (read), operation reservation, then quota ledger.
+// When Object Lock already holds the metadata transaction read lock, reuse
+// that guard: reacquiring it behind a waiting metadata writer would deadlock.
 
 #[cfg(not(any(test, feature = "test-util")))]
 const ORPHAN_MIN_AGE_SECONDS: i64 = 30;
@@ -211,7 +213,7 @@ pub(crate) struct QuotaContext {
     capability_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     snapshot_admission: Option<QuotaAdmission>,
     legacy_data_movement: bool,
-    metadata_guard: Option<NamespaceLockGuard>,
+    metadata_guard: Option<Arc<NamespaceLockGuard>>,
     pool_index: Option<usize>,
     set_index: Option<usize>,
 }
@@ -296,7 +298,7 @@ impl QuotaContext {
                     limit: quota_limit,
                 });
             }
-            if operation_guard.is_lock_lost() || metadata_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) {
+            if operation_guard.is_lock_lost() || metadata_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
                 return Err(StorageError::NamespaceLockQuorumUnavailable {
                     mode: "quota_reservation",
                     bucket: ledger_data.bucket.clone(),
@@ -333,7 +335,7 @@ struct LedgerReservationData {
 pub(crate) struct QuotaReservation {
     ledger: Option<LedgerReservationData>,
     operation_guard: Option<NamespaceLockGuard>,
-    metadata_guard: Option<NamespaceLockGuard>,
+    metadata_guard: Option<Arc<NamespaceLockGuard>>,
     capability_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     state: ReservationState,
 }
@@ -347,7 +349,7 @@ enum ReservationState {
 }
 
 impl QuotaReservation {
-    fn unlimited(metadata_guard: Option<NamespaceLockGuard>) -> Self {
+    fn unlimited(metadata_guard: Option<Arc<NamespaceLockGuard>>) -> Self {
         Self {
             ledger: None,
             operation_guard: None,
@@ -359,7 +361,7 @@ impl QuotaReservation {
 
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.operation_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
-            || self.metadata_guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
+            || self.metadata_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
     }
 
     pub(crate) fn capability_proof_matches(&self) -> bool {
@@ -439,11 +441,12 @@ pub(crate) async fn begin(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
     object: &str,
-    snapshot_admission: Option<QuotaAdmission>,
-    data_movement: bool,
+    opts: &ObjectOptions,
     pool_index: usize,
     set_index: usize,
 ) -> Result<QuotaContext> {
+    let snapshot_admission = opts.quota_admission;
+    let data_movement = opts.data_movement;
     if crate::bucket::utils::is_meta_bucketname(bucket) {
         return Ok(QuotaContext {
             store: None,
@@ -498,7 +501,7 @@ pub(crate) async fn begin(
         });
     }
 
-    let metadata_guard = metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(ctx, bucket).await?;
+    let metadata_guard = metadata_sys::acquire_bucket_metadata_transaction_read_lock_for_options_in(ctx, bucket, opts).await?;
     let (quota, bucket_incarnation, quota_revision) =
         metadata_sys::get_quota_config_and_incarnation_from_disk_in(ctx, bucket).await?;
     if metadata_guard.is_lock_lost() {

crates/ecstore/src/core/pools_test.rs

@@ -481,6 +481,9 @@ mod decommission_lock_order_tests {
         if let Some(deployment_id) = other_store.ctx.deployment_id() {
             ctx.set_deployment_id(deployment_id);
         }
+        for pool in &mut pools {
+            Arc::make_mut(pool).set_instance_ctx_for_test(Arc::clone(&ctx));
+        }
         let store = Arc::new(crate::store::ECStore {
             id: uuid::Uuid::new_v4(),
             disk_map: other_store.disk_map.clone(),
@@ -6645,7 +6648,19 @@ mod decommission_lock_order_tests {
 
         let restore_get_barrier = if matches!(mutation, ExternalObjectMutation::Restore) {
             let tier_name = format!("ORDERRESTORE{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-            let backend = register_mock_tier(&store.pools[2].instance_ctx().tier_config_mgr(), &tier_name).await;
+            let source_tiers = store.tier_config_mgr();
+            let restore_tiers = other_store.tier_config_mgr();
+            let backend = register_mock_tier(&source_tiers, &tier_name).await;
+            // Both peers must resolve the same persisted backend identity, including its prefix.
+            let tier_config = source_tiers.read().await.tiers[&tier_name].clone();
+            restore_tiers.write().await.tiers.insert(tier_name.clone(), tier_config);
+            crate::services::tier::tier::TierConfigMgr::install_test_driver_in(
+                &restore_tiers,
+                &tier_name,
+                Box::new(backend.clone()),
+            )
+            .await
+            .expect("install the same mock tier on the restoring peer");
             store.pools[2]
                 .transition_object(
                     &bucket,
@@ -6919,7 +6934,10 @@ mod decommission_lock_order_tests {
             }
         });
         if let Some(get_barrier) = restore_get_barrier.as_ref() {
-            get_barrier.wait_until_paused().await;
+            tokio::select! {
+                () = get_barrier.wait_until_paused() => {}
+                result = &mut ordinary_mutation => panic!("restore finished before the tier GET barrier: {result:?}"),
+            }
             let read_opts = ObjectOptions {
                 skip_decommissioned: true,
                 ..Default::default()

crates/ecstore/src/core/sets.rs

@@ -324,6 +324,15 @@ impl Sets {
         &self.ctx
     }
 
+    /// Keep simulated peers' metadata ownership separate while sharing disks and lock clients.
+    #[cfg(test)]
+    pub(crate) fn set_instance_ctx_for_test(&mut self, ctx: Arc<InstanceContext>) {
+        for set in &mut self.disk_set {
+            Arc::make_mut(set).set_instance_ctx_for_test(Arc::clone(&ctx));
+        }
+        self.ctx = ctx;
+    }
+
     async fn monitor_and_connect_endpoints_task(sets: Weak<Sets>, mut rx: Receiver<()>) {
         let startup_delay = tokio::time::sleep(Duration::from_secs(5));
         tokio::pin!(startup_delay);

crates/ecstore/src/object_api/types.rs

@@ -41,7 +41,7 @@ impl Debug for NamespaceLockFence {
 }
 
 impl NamespaceLockFence {
-    fn new() -> Self {
+    pub(crate) fn new() -> Self {
         Self {
             signals: Arc::default(),
             #[cfg(test)]
@@ -153,7 +153,7 @@ pub struct ObjectLockConfigSnapshot {
     state: crate::bucket::metadata_sys::ObjectLockConfigState,
     lifecycle_fence: NamespaceLockFence,
     _lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
-    metadata_transaction_guard: Option<rustfs_lock::NamespaceLockGuard>,
+    metadata_transaction_guard: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
 }
 
 impl ObjectLockConfigSnapshot {
@@ -170,6 +170,7 @@ impl ObjectLockConfigSnapshot {
         }
     }
 
+    #[cfg(test)]
     pub(crate) fn for_store_bucket(
         store_id: Uuid,
         bucket: &str,
@@ -210,7 +211,7 @@ impl ObjectLockConfigSnapshot {
             state,
             lifecycle_fence,
             _lifecycle_guard: Some(lifecycle_guard),
-            metadata_transaction_guard: Some(metadata_transaction_guard),
+            metadata_transaction_guard: Some(Arc::new(metadata_transaction_guard)),
         }
     }
 
@@ -221,7 +222,7 @@ impl ObjectLockConfigSnapshot {
         config_revision: OffsetDateTime,
         state: crate::bucket::metadata_sys::ObjectLockConfigState,
         lifecycle_fence: NamespaceLockFence,
-        metadata_transaction_guard: rustfs_lock::NamespaceLockGuard,
+        metadata_transaction_guard: Arc<rustfs_lock::NamespaceLockGuard>,
     ) -> Self {
         Self {
             store_id: Some(store_id),
@@ -265,6 +266,23 @@ impl ObjectLockConfigSnapshot {
                 .is_some_and(|guard| !guard.is_lock_lost())
     }
 
+    /// Share the held transaction lock without queuing another reader behind
+    /// a metadata writer that is itself waiting for this snapshot to drop.
+    pub(crate) fn metadata_transaction_guard_for(
+        &self,
+        store_id: Uuid,
+        bucket: &str,
+        expected_incarnation_id: Option<Uuid>,
+    ) -> Option<Arc<rustfs_lock::NamespaceLockGuard>> {
+        let bucket_incarnation_id = self.bucket_incarnation_id?;
+        if expected_incarnation_id.is_some_and(|expected| expected != bucket_incarnation_id)
+            || !self.is_valid_for_destructive_put(store_id, bucket, bucket_incarnation_id)
+        {
+            return None;
+        }
+        self.metadata_transaction_guard.clone()
+    }
+
     pub(crate) fn add_lock_fences(&self, opts: &mut ObjectOptions) {
         opts.bucket_lifecycle_lock_fence
             .get_or_insert_with(NamespaceLockFence::new)

crates/ecstore/src/services/rebalance/mod.rs

@@ -222,10 +222,14 @@ async fn test_pool_stores_with_contexts(
         std::sync::Arc::clone(&ctx)
     };
     let make_store = |store_ctx: std::sync::Arc<crate::runtime::instance::InstanceContext>| {
+        let mut store_pools = pools.clone();
+        for pool in &mut store_pools {
+            std::sync::Arc::make_mut(pool).set_instance_ctx_for_test(std::sync::Arc::clone(&store_ctx));
+        }
         std::sync::Arc::new(crate::store::ECStore {
             id: uuid::Uuid::new_v4(),
             disk_map: std::collections::HashMap::new(),
-            pools: pools.clone(),
+            pools: store_pools,
             peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, std::sync::Arc::clone(&store_ctx)),
             pool_meta: tokio::sync::RwLock::new(pool_meta.clone()),
             rebalance_meta: tokio::sync::RwLock::new(rebalance_meta.clone()),

crates/ecstore/src/set_disk/mod.rs

@@ -4703,6 +4703,12 @@ impl SetDisks {
         &self.ctx
     }
 
+    #[cfg(test)]
+    pub(crate) fn set_instance_ctx_for_test(&mut self, ctx: Arc<InstanceContext>) {
+        self.local_lock_manager = ctx.lock_manager();
+        self.ctx = ctx;
+    }
+
     /// Read the persisted bucket identity through this set's metadata owner.
     /// Missing or non-authoritative legacy identities remain errors.
     pub async fn bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Uuid> {

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

@@ -2613,16 +2613,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         fi.parts = Vec::with_capacity(uploaded_parts.len());
 
-        let quota_context = reservation::begin(
-            &self.ctx,
-            bucket,
-            object,
-            opts.quota_admission,
-            opts.data_movement,
-            self.pool_index,
-            self.set_index,
-        )
-        .await?;
+        let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?;
         let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
         let preserve_replication_ciphertext = opts.replication_request
             && contains_key_str(&fi.metadata, rustfs_utils::http::SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT);

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

@@ -3427,8 +3427,8 @@ impl SetDisks {
         Ok(removed)
     }
 
-    async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid) -> Result<()> {
-        let current = metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+    async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid, opts: &ObjectOptions) -> Result<()> {
+        let current = metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, opts).await?;
         if current != expected {
             return Err(StorageError::BucketNotFound(bucket.to_string()));
         }
@@ -4234,16 +4234,7 @@ impl SetDisks {
                 None
             };
 
-            let quota_context = reservation::begin(
-                &self.ctx,
-                bucket,
-                object,
-                opts.quota_admission,
-                opts.data_movement,
-                self.pool_index,
-                self.set_index,
-            )
-            .await?;
+            let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?;
             let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
             let mut replication_quota_size = None;
 
@@ -8869,7 +8860,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 return Ok(ObjectInfo::default());
             }
             if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id {
-                self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+                self.validate_bucket_incarnation(bucket, expected_incarnation_id, &opts)
+                    .await?;
             }
             ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
             begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
@@ -9283,7 +9275,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         check_object_lock_retention_update(bucket, object, &obj_info, opts)?;
 
         if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id {
-            self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+            self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts)
+                .await?;
         }
         if _lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
             || opts
@@ -9863,7 +9856,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 .await?
                 .acquire_bucket_lifecycle_read_lock(bucket)
                 .await?;
-            self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?;
+            self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts)
+                .await?;
             Some(guard)
         } else {
             None

crates/ecstore/src/store/mod.rs

@@ -395,12 +395,19 @@ async fn scan_metadata_less_residue_with_budget(
     Ok(scan)
 }
 
-async fn enqueue_transition_after_write(result: Result<ObjectInfo>, src: LcEventSrc) -> Result<ObjectInfo> {
+async fn enqueue_transition_after_write(
+    store: &ECStore,
+    result: Result<ObjectInfo>,
+    src: LcEventSrc,
+    opts: &ObjectOptions,
+) -> Result<ObjectInfo> {
     match result {
         Ok(oi) => {
             if should_enqueue_transition_immediately(&oi) {
                 enqueue_transition_immediate(&oi, src.clone()).await;
-                enqueue_immediate_expiry(&oi, src).await;
+                if let Ok(api) = metadata_sys::object_store_in(&store.ctx).await {
+                    enqueue_immediate_expiry(api, &oi, src, opts).await;
+                }
             }
             Ok(oi)
         }
@@ -1281,9 +1288,11 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
         let result = match self.handle_put_object(bucket, object, data, opts).await {
-            Ok((object_info, old_current_size)) => enqueue_transition_after_write(Ok(object_info), LcEventSrc::S3PutObject)
-                .await
-                .map(|object_info| (object_info, old_current_size)),
+            Ok((object_info, old_current_size)) => {
+                enqueue_transition_after_write(self, Ok(object_info), LcEventSrc::S3PutObject, opts)
+                    .await
+                    .map(|object_info| (object_info, old_current_size))
+            }
             Err(err) => Err(err),
         };
         if result.is_ok() {
@@ -1356,9 +1365,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
         dst_opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         let result = enqueue_transition_after_write(
+            self,
             self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
                 .await,
             LcEventSrc::S3CopyObject,
+            dst_opts,
         )
         .await;
         if result.is_ok() {
@@ -1625,10 +1636,12 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for ECStore {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         let result = enqueue_transition_after_write(
+            self.as_ref(),
             self.clone()
                 .handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
                 .await,
             LcEventSrc::S3CompleteMultipartUpload,
+            opts,
         )
         .await;
         if result.is_ok() {

crates/ecstore/src/store/multipart.rs

@@ -355,7 +355,7 @@ impl ECStore {
         if let Some(guard) = guard.as_ref() {
             opts.add_bucket_lifecycle_lock_guard(guard);
         }
-        let current = crate::bucket::metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+        let current = crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
         if opts.expected_bucket_incarnation_id != Some(current) {
             return Err(StorageError::BucketNotFound(bucket.to_string()));
         }
@@ -1047,6 +1047,7 @@ impl ECStore {
         opts: &ObjectOptions,
         publication_fence: Option<RemoteTuplePublicationFence>,
     ) -> Result<ObjectInfo> {
+        let request_opts = opts;
         let (target_pool_idx, mutation_fence) = target;
         check_complete_multipart_args(bucket, object, upload_id)?;
         if !opts.data_movement {
@@ -1132,7 +1133,8 @@ impl ECStore {
             )
             .await;
         drop(publication_guard);
-        let result = enqueue_transition_after_write(result, LcEventSrc::S3CompleteMultipartUpload).await;
+        let result =
+            enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, request_opts).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self.as_ref(), bucket).await;
         }

crates/ecstore/src/store/object.rs

@@ -29,8 +29,8 @@ use crate::bucket::lifecycle::{
     },
 };
 use crate::bucket::metadata_sys::{
-    acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_in, get_cached_bucket_incarnation_id_in,
-    get_object_lock_config_and_incarnation_from_disk_in,
+    acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_for_options_in,
+    get_cached_bucket_incarnation_id_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,
@@ -2837,7 +2837,7 @@ impl ECStore {
             config_revision,
             state,
             lifecycle_fence.clone(),
-            metadata_guard,
+            Arc::new(metadata_guard),
         )))
     }
 
@@ -4296,6 +4296,7 @@ impl ECStore {
         if !opts.data_movement {
             return Err(Error::other("data movement PUT requires data_movement options"));
         }
+        let request_opts = opts;
         let (object, mut opts) = self.prepare_put_object(bucket, object, opts).await?;
         ensure_decommission_capacity_mutation_id(bucket, &object, &mut opts);
         let idx = self
@@ -4338,7 +4339,7 @@ impl ECStore {
                 },
             )
             .await;
-        let result = enqueue_transition_after_write(result, LcEventSrc::S3PutObject).await;
+        let result = enqueue_transition_after_write(self, result, LcEventSrc::S3PutObject, request_opts).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self, bucket).await;
         }
@@ -4445,7 +4446,7 @@ impl ECStore {
         };
         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_in(&self.ctx, dst_bucket).await?;
+            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)
@@ -4826,22 +4827,26 @@ impl ECStore {
             get_cached_bucket_incarnation_id_in(&self.ctx, bucket).await?;
         }
         let _object_lock_metadata_guard = if !is_meta_bucketname(bucket) {
-            Some(acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?)
+            Some(Arc::new(acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?))
         } else {
             None
         };
         if let Some(guard) = _object_lock_metadata_guard.as_ref() {
             opts.add_namespace_lock_guard(guard);
         }
-        let current_bucket_incarnation_id = if _object_lock_metadata_guard.is_some() {
+        let current_bucket_incarnation_id = if let Some(metadata_guard) = _object_lock_metadata_guard.as_ref() {
             let (state, incarnation_id, config_revision) =
                 get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await?;
-            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket(
+            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket_under_lifecycle_fence(
                 self.id,
                 bucket,
                 incarnation_id,
                 config_revision,
                 state,
+                opts.bucket_lifecycle_lock_fence
+                    .clone()
+                    .unwrap_or_else(NamespaceLockFence::new),
+                Arc::clone(metadata_guard),
             )));
             Some(incarnation_id)
         } else {
@@ -5260,26 +5265,32 @@ impl ECStore {
         let _object_lock_metadata_guard = if is_meta_bucketname(bucket) {
             None
         } else {
-            Some(match acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await {
-                Ok(guard) => guard,
-                Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
-            })
+            Some(Arc::new(
+                match acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await {
+                    Ok(guard) => guard,
+                    Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
+                },
+            ))
         };
         if let Some(guard) = _object_lock_metadata_guard.as_ref() {
             opts.add_namespace_lock_guard(guard);
         }
-        let current_bucket_incarnation_id = if _object_lock_metadata_guard.is_some() {
+        let current_bucket_incarnation_id = if let Some(metadata_guard) = _object_lock_metadata_guard.as_ref() {
             let (state, incarnation_id, config_revision) =
                 match get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await {
                     Ok(snapshot) => snapshot,
                     Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
                 };
-            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket(
+            opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket_under_lifecycle_fence(
                 self.id,
                 bucket,
                 incarnation_id,
                 config_revision,
                 state,
+                opts.bucket_lifecycle_lock_fence
+                    .clone()
+                    .unwrap_or_else(NamespaceLockFence::new),
+                Arc::clone(metadata_guard),
             )));
             Some(incarnation_id)
         } else {
@@ -5620,7 +5631,7 @@ impl ECStore {
             opts.add_bucket_lifecycle_lock_guard(guard);
         }
         if !is_meta_bucketname(bucket) {
-            let current_incarnation_id = get_bucket_incarnation_id_in(&self.ctx, bucket).await?;
+            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()));
             }
@@ -5693,7 +5704,7 @@ impl ECStore {
             None
         } else {
             let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
-            let current_incarnation_id = get_bucket_incarnation_id_in(&self.ctx, 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)

crates/ecstore/tests/metadata_read_lock_reuse_test.rs

@@ -0,0 +1,288 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![cfg(feature = "test-util")]
+
+mod storage_api;
+
+use std::sync::Arc;
+use std::time::Duration;
+use storage_api::metadata_lock::{
+    BucketOperations, CompletePart, Error, MakeBucketOptions, MultipartOperations, NamespaceLocking, ObjectIO, ObjectOperations,
+    ObjectOptions, PutObjReader, PutObjectCommitBarrier, PutObjectCommitPause, init_bucket_metadata_sys,
+    isolated_store_over_temp_disks,
+};
+use tokio::io::AsyncReadExt;
+use tokio::time::timeout;
+use uuid::Uuid;
+
+#[tokio::test]
+async fn replica_put_completes_while_metadata_writer_waits_for_its_snapshot() {
+    replica_write_with_waiting_metadata_writer(false).await;
+}
+
+#[tokio::test]
+async fn replica_multipart_completes_while_metadata_writer_waits_for_its_snapshot() {
+    replica_write_with_waiting_metadata_writer(true).await;
+}
+
+async fn replica_write_with_waiting_metadata_writer(multipart: bool) {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "replica-metadata-read-reuse";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    store
+        .update_bucket_metadata_config(
+            bucket,
+            "lifecycle.xml",
+            br#"<LifecycleConfiguration><Rule><ID>expire-old-versions</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule></LifecycleConfiguration>"#.to_vec(),
+        )
+        .await
+        .expect("configure post-write lifecycle evaluation");
+    let opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        replication_request: true,
+        expected_bucket_incarnation_id: Some(store.bucket_incarnation_id(bucket).await.expect("load incarnation")),
+        object_lock_config_snapshot: Some(
+            store
+                .object_lock_config_snapshot(bucket)
+                .await
+                .expect("capture Object Lock snapshot"),
+        ),
+        ..Default::default()
+    };
+    let upload = if multipart {
+        let upload = store
+            .new_multipart_upload(bucket, "object", &opts)
+            .await
+            .expect("stage upload");
+        let mut data = PutObjReader::from_vec(b"replicated version".to_vec());
+        let part = store
+            .put_object_part(bucket, "object", &upload.upload_id, 1, &mut data, &opts)
+            .await
+            .expect("stage part");
+        Some((
+            upload.upload_id,
+            vec![CompletePart {
+                part_num: 1,
+                etag: part.etag,
+                ..Default::default()
+            }],
+        ))
+    } else {
+        None
+    };
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create metadata writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    // Leave the writer registered while the replica continues. A second read
+    // acquisition would queue behind this writer, which needs our first read.
+    assert!(futures::poll!(&mut writer).is_pending());
+    timeout(Duration::from_secs(2), async {
+        if let Some((upload_id, parts)) = upload {
+            store
+                .clone()
+                .complete_multipart_upload(bucket, "object", &upload_id, parts, &opts)
+                .await
+        } else {
+            let mut data = PutObjReader::from_vec(b"replicated version".to_vec());
+            store.put_object(bucket, "object", &mut data, &opts).await
+        }
+    })
+    .await
+    .expect("replica must reuse its metadata read lock instead of waiting behind the writer")
+    .expect("replica PUT should commit");
+    assert!(futures::poll!(&mut writer).is_pending(), "the snapshot must still fence metadata updates");
+    let read_opts = ObjectOptions {
+        version_id: opts.version_id.clone(),
+        ..Default::default()
+    };
+    drop(opts);
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("metadata writer must proceed after the snapshot drops")
+        .expect("acquire metadata writer");
+    let mut reader = store
+        .get_object_reader(bucket, "object", None, Default::default(), &read_opts)
+        .await
+        .expect("read committed replica version");
+    let mut bytes = Vec::new();
+    reader.stream.read_to_end(&mut bytes).await.expect("read replica body");
+    assert_eq!(bytes, b"replicated version");
+}
+
+#[tokio::test]
+async fn reused_metadata_lock_still_enforces_quota() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-quota";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    store
+        .update_bucket_metadata_config(bucket, "quota.json", br#"{"quota":1}"#.to_vec())
+        .await
+        .expect("set quota");
+    let mut opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        expected_bucket_incarnation_id: Some(store.bucket_incarnation_id(bucket).await.expect("load incarnation")),
+        object_lock_config_snapshot: Some(store.object_lock_config_snapshot(bucket).await.expect("capture snapshot")),
+        ..Default::default()
+    };
+    assert!(opts.set_quota_admission(0, 1));
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    assert!(futures::poll!(&mut writer).is_pending());
+    let mut data = PutObjReader::from_vec(b"too large".to_vec());
+    let error = timeout(Duration::from_secs(2), store.put_object(bucket, "object", &mut data, &opts))
+        .await
+        .expect("quota admission must not reacquire the lock")
+        .expect_err("quota must reject growth");
+    assert!(matches!(error, Error::QuotaExceeded { limit: 1, .. }), "{error:?}");
+    drop(opts);
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("release metadata reader after denial")
+        .expect("acquire writer");
+}
+
+#[tokio::test]
+async fn cancelled_put_releases_shared_metadata_guard() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-cancel";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let opts = ObjectOptions {
+        versioned: true,
+        version_id: Some(Uuid::new_v4().to_string()),
+        object_lock_config_snapshot: Some(store.object_lock_config_snapshot(bucket).await.expect("capture snapshot")),
+        ..Default::default()
+    };
+    let barrier = PutObjectCommitBarrier::install(bucket, "object", PutObjectCommitPause::AfterQuotaReservation);
+    let put_store = Arc::clone(&store);
+    let put = tokio::spawn(async move {
+        let mut data = PutObjReader::from_vec(b"cancel before commit".to_vec());
+        put_store.put_object(bucket, "object", &mut data, &opts).await
+    });
+    barrier.wait_until_paused().await;
+    let lock = store
+        .new_ns_lock(".rustfs.sys", &format!("bucket-targets/{bucket}/transaction.lock"))
+        .await
+        .expect("create writer");
+    let writer = lock.get_write_lock(Duration::from_secs(5));
+    tokio::pin!(writer);
+    assert!(futures::poll!(&mut writer).is_pending());
+    put.abort();
+    assert!(put.await.expect_err("PUT should be cancelled").is_cancelled());
+    timeout(Duration::from_secs(2), writer)
+        .await
+        .expect("all metadata reader owners must drop after cancellation")
+        .expect("acquire writer");
+}
+
+#[tokio::test]
+async fn multipart_rejects_metadata_snapshots_from_another_scope() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let (_other_dirs, other_store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&other_store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-scope";
+    for target in [&store, &other_store] {
+        target
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+    }
+    store
+        .make_bucket("other-bucket", &MakeBucketOptions::default())
+        .await
+        .expect("create other bucket");
+    let other_store_incarnation = other_store
+        .bucket_incarnation_id(bucket)
+        .await
+        .expect("load other store incarnation");
+    let other_bucket_incarnation = store
+        .bucket_incarnation_id("other-bucket")
+        .await
+        .expect("load other bucket incarnation");
+    for (snapshot_store, snapshot_bucket, expected) in [
+        (&other_store, bucket, other_store_incarnation),
+        (&store, "other-bucket", other_bucket_incarnation),
+        (&store, bucket, Uuid::new_v4()),
+    ] {
+        let opts = ObjectOptions {
+            expected_bucket_incarnation_id: Some(expected),
+            object_lock_config_snapshot: Some(
+                snapshot_store
+                    .object_lock_config_snapshot(snapshot_bucket)
+                    .await
+                    .expect("capture snapshot"),
+            ),
+            ..Default::default()
+        };
+        let error = store
+            .new_multipart_upload(bucket, "object", &opts)
+            .await
+            .expect_err("foreign snapshot must be rejected");
+        assert!(error.to_string().contains("valid metadata transaction fence"), "{error:?}");
+    }
+}
+
+#[tokio::test]
+async fn delete_prefix_reuses_its_metadata_snapshot_for_generation_validation() {
+    let (_dirs, store) = isolated_store_over_temp_disks().await;
+    init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+    let bucket = "metadata-read-reuse-delete";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let mut data = PutObjReader::from_vec(b"remove prefix".to_vec());
+    store
+        .put_object(bucket, "prefix/object", &mut data, &ObjectOptions::default())
+        .await
+        .expect("seed object");
+    store
+        .delete_object(
+            bucket,
+            "prefix/",
+            ObjectOptions {
+                delete_prefix: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("delete prefix with an internally captured metadata snapshot");
+    let error = store
+        .get_object_info(bucket, "prefix/object", &ObjectOptions::default())
+        .await
+        .expect_err("prefix object must be deleted");
+    assert!(matches!(error, Error::ObjectNotFound(..)), "{error:?}");
+}

crates/ecstore/tests/storage_api.rs

@@ -6,6 +6,17 @@ pub(crate) use rustfs_ecstore::api::erasure::Erasure;
 pub(crate) use rustfs_ecstore::api::object::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
 pub(crate) use rustfs_ecstore::api::{error::Error, set_disk::SetDisks, storage::ECStore};
 use rustfs_storage_api as storage_contracts;
+#[cfg(feature = "test-util")]
+pub(crate) mod metadata_lock {
+    pub(crate) use super::storage_contracts::{
+        BucketOperations, CompletePart, MakeBucketOptions, MultipartOperations, NamespaceLocking, ObjectIO, ObjectOperations,
+    };
+    pub(crate) use super::{Error, ObjectOptions, PutObjReader};
+    pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{
+        init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks,
+    };
+    pub(crate) use rustfs_ecstore::api::set_disk::test_util::{PutObjectCommitBarrier, PutObjectCommitPause};
+}
 
 pub(crate) mod contract_compat {
     pub(crate) use super::storage_contracts::{