From cd10981132b42a6c25ae0b4b5aa5efe302e89c8c Mon Sep 17 00:00:00 2001 From: Jason Kossis Date: Wed, 16 Sep 2026 18:39:25 -0400 Subject: [PATCH] fix(ecstore): reuse metadata read guards during writes (#7941) --- .config/nextest.toml | 10 + .../bucket/lifecycle/bucket_lifecycle_ops.rs | 10 +- .../src/bucket/lifecycle/metadata_boundary.rs | 10 +- crates/ecstore/src/bucket/metadata_sys.rs | 42 ++- .../ecstore/src/bucket/quota/reservation.rs | 19 +- crates/ecstore/src/core/pools_test.rs | 22 +- crates/ecstore/src/core/sets.rs | 9 + crates/ecstore/src/object_api/types.rs | 26 +- crates/ecstore/src/services/rebalance/mod.rs | 6 +- crates/ecstore/src/set_disk/mod.rs | 6 + crates/ecstore/src/set_disk/ops/multipart.rs | 11 +- crates/ecstore/src/set_disk/ops/object.rs | 24 +- crates/ecstore/src/store/mod.rs | 23 +- crates/ecstore/src/store/multipart.rs | 6 +- crates/ecstore/src/store/object.rs | 43 ++- .../tests/metadata_read_lock_reuse_test.rs | 288 ++++++++++++++++++ crates/ecstore/tests/storage_api.rs | 11 + 17 files changed, 495 insertions(+), 71 deletions(-) create mode 100644 crates/ecstore/tests/metadata_read_lock_reuse_test.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 3fa44792f..5e5b5cd33 100644 --- a/.config/nextest.toml +++ b/.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]] diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 048ba1bfa..8c4ac1322 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/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, 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(&( diff --git a/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs b/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs index 2bb4941fb..865c34322 100644 --- a/crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs +++ b/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 { - 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 { + 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 ecb2ba250..06bda6f19 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/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 { + 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> { + 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>, 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 { 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 { diff --git a/crates/ecstore/src/bucket/quota/reservation.rs b/crates/ecstore/src/bucket/quota/reservation.rs index d4a35c23e..7157d3262 100644 --- a/crates/ecstore/src/bucket/quota/reservation.rs +++ b/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, snapshot_admission: Option, legacy_data_movement: bool, - metadata_guard: Option, + metadata_guard: Option>, pool_index: Option, set_index: Option, } @@ -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, operation_guard: Option, - metadata_guard: Option, + metadata_guard: Option>, capability_proof: Option, state: ReservationState, } @@ -347,7 +349,7 @@ enum ReservationState { } impl QuotaReservation { - fn unlimited(metadata_guard: Option) -> Self { + fn unlimited(metadata_guard: Option>) -> 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, - data_movement: bool, + opts: &ObjectOptions, pool_index: usize, set_index: usize, ) -> Result { + 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() { diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 2c08f25a7..6fc60bb04 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/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() diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index a83c4d699..01d2e0864 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/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) { + 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, mut rx: Receiver<()>) { let startup_delay = tokio::time::sleep(Duration::from_secs(5)); tokio::pin!(startup_delay); diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index eb0851585..37ffc6ff2 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/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, - metadata_transaction_guard: Option, + metadata_transaction_guard: Option>, } 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, ) -> 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, + ) -> Option> { + 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) diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 352d765e0..1fb063237 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/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| { + 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()), diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index fda7283b4..0fd0cb6be 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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) { + 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 { diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 54234239e..15a4603f2 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/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); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 3b0e0decd..c5daef759 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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 diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 2198f7d2b..625cf32a2 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/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, src: LcEventSrc) -> Result { +async fn enqueue_transition_after_write( + store: &ECStore, + result: Result, + src: LcEventSrc, + opts: &ObjectOptions, +) -> Result { 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)> { 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 { 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 { 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() { diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 0cd7dd837..e0ce0e79c 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/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, ) -> Result { + 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; } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index a801ffea1..f138d06d0 100644 --- a/crates/ecstore/src/store/object.rs +++ b/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) diff --git a/crates/ecstore/tests/metadata_read_lock_reuse_test.rs b/crates/ecstore/tests/metadata_read_lock_reuse_test.rs new file mode 100644 index 000000000..6433bf0cd --- /dev/null +++ b/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#"expire-old-versionsEnabled1"#.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:?}"); +} diff --git a/crates/ecstore/tests/storage_api.rs b/crates/ecstore/tests/storage_api.rs index 1b2005dee..32e33be66 100644 --- a/crates/ecstore/tests/storage_api.rs +++ b/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::{