diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 0d3caa90c..860e31fb0 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -122,13 +122,13 @@ pub mod bucket { pub mod metadata_sys { pub use crate::bucket::metadata_sys::{ - BucketMetadataSys, acquire_bucket_targets_transaction_lock, delete, get, get_accelerate_config, get_bucket_policy, + BucketMetadataSys, acquire_bucket_metadata_transaction_lock, delete, get, get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config, get_object_lock_config, get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, - update, update_bucket_targets_under_transaction_lock, update_config_with, + update, update_bucket_targets_under_transaction_lock, update_config_with, update_under_transaction_lock, }; } diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 9896b0ea7..6dfce351e 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -242,73 +242,138 @@ pub(crate) async fn remove_bucket_metadata_in(ctx: &crate::runtime::instance::In Ok(lock.remove(bucket).await) } +/// Rewrite one config file of a bucket's metadata, serialized cluster-wide. +/// +/// See [`acquire_bucket_metadata_transaction_lock`] for why every config +/// write — not just the replication-targets one — has to hold that lock. pub async fn update(bucket: &str, config_file: &str, data: Vec) -> Result { - let bucket_meta_sys_lock = get_bucket_metadata_sys()?; - let _targets_guard = if config_file == BUCKET_TARGETS_FILE { - Some(acquire_bucket_targets_transaction_lock(bucket).await?) - } else { - None - }; - let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; + update_with_sys(get_bucket_metadata_sys()?, bucket, config_file, data).await +} +pub async fn delete(bucket: &str, config_file: &str) -> Result { + delete_with_sys(get_bucket_metadata_sys()?, bucket, config_file).await +} + +/// [`update`] against an explicitly supplied metadata system. +/// +/// The free functions resolve the instance's own system; this variant takes +/// it as an argument so a test can drive two independent systems over one +/// backing store — the in-process stand-in for two nodes, which is the only +/// configuration where the transaction lock is what does the serializing. +async fn update_with_sys( + sys: Arc>, + bucket: &str, + config_file: &str, + data: Vec, +) -> Result { + let (_transaction_guard, mut sys) = acquire_config_write_guards(sys, bucket).await?; + sys.update(bucket, config_file, data).await +} + +/// [`delete`] against an explicitly supplied metadata system. See +/// [`update_with_sys`]. +async fn delete_with_sys(sys: Arc>, bucket: &str, config_file: &str) -> Result { + let (_transaction_guard, mut sys) = acquire_config_write_guards(sys, bucket).await?; + sys.delete(bucket, config_file).await +} + +/// Take, in the one order every config write uses, the two guards a +/// read-modify-write of a bucket's metadata needs: the cluster-wide +/// transaction lock first, then this process's metadata-system write guard. +/// +/// The order is load-bearing. Acquiring the process-local guard first would +/// park every local reader and writer of *every* bucket behind a lock whose +/// holder may be another node, turning remote contention into a local stall. +async fn acquire_config_write_guards( + sys: Arc>, + bucket: &str, +) -> Result<(rustfs_lock::NamespaceLockGuard, tokio::sync::OwnedRwLockWriteGuard)> { + let transaction_guard = acquire_transaction_lock_with_sys(&sys, bucket).await?; + let sys_guard = sys.write_owned().await; + Ok((transaction_guard, sys_guard)) +} + +/// Rewrite one config file while the caller already holds this bucket's +/// transaction lock. +/// +/// [`update`] would deadlock here: the lock is not reentrant, so a holder +/// that called it would block until its own guard timed out. +pub async fn update_under_transaction_lock(bucket: &str, config_file: &str, data: Vec) -> Result { + let bucket_meta_sys_lock = get_bucket_metadata_sys()?; + let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; bucket_meta_sys.update(bucket, config_file, data).await } pub async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result { - let bucket_meta_sys_lock = get_bucket_metadata_sys()?; - let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; - bucket_meta_sys.update(bucket, BUCKET_TARGETS_FILE, data).await + update_under_transaction_lock(bucket, BUCKET_TARGETS_FILE, data).await } -/// Read-modify-write one bucket config file under the metadata system's -/// outer write guard. +/// Read-modify-write one bucket config file under both guards a config +/// write takes. /// /// `mutate` sees the freshly loaded on-disk metadata and returns the /// replacement payload for `config_file` (empty clears it, like /// [`delete`]). Both the read and the persisted write happen inside the -/// same guard that [`update`] uses, so within this process the rewrite can -/// neither clobber a concurrent update to another config file nor lose a -/// concurrent write to the same one — unlike caching a mutated clone of -/// previously read metadata. +/// same guards [`update`] takes, so the rewrite can neither clobber a +/// concurrent update to another config file nor lose a concurrent write to +/// the same one — unlike caching a mutated clone of previously read +/// metadata. /// -/// This guard is process-local. Writers on other nodes still race, exactly -/// as they do for [`update`]: each rewrites the whole metadata file, so the -/// later save wins. What this narrows is the window — from "as stale as the -/// local cache" down to a single metadata read plus write. +/// That exclusion is cluster-wide, not merely process-local: the transaction +/// lock is now taken for every config file rather than only the replication +/// targets one, so a writer on another node cannot land a whole-file save in +/// the middle of this read-modify-write. pub async fn update_config_with(bucket: &str, config_file: &str, mutate: F) -> Result where F: FnOnce(&BucketMetadata) -> Result> + Send, { - let bucket_meta_sys_lock = get_bucket_metadata_sys()?; - let _targets_guard = if config_file == BUCKET_TARGETS_FILE { - Some(acquire_bucket_targets_transaction_lock(bucket).await?) - } else { - None - }; - let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; - bucket_meta_sys.update_config_with(bucket, config_file, mutate).await + let (_transaction_guard, mut sys) = acquire_config_write_guards(get_bucket_metadata_sys()?, bucket).await?; + sys.update_config_with(bucket, config_file, mutate).await } -pub async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { - let bucket_meta_sys_lock = get_bucket_metadata_sys()?; - let api = bucket_meta_sys_lock.read().await.object_store(); +/// Acquire a bucket's metadata transaction lock, held across a whole +/// read-modify-write of its metadata file. +/// +/// Every config write loads the entire [`BucketMetadata`] blob, replaces one +/// field, and saves the whole thing back. The namespace locks inside +/// `read_config`/`save_config` are taken and released separately, so they do +/// not span that cycle: two nodes updating *different* config files of one +/// bucket both load the same blob, each set their own field, and the later +/// save drops the other's — with both clients already told 2xx. This is not +/// last-writer-wins on one document; an orthogonal config silently vanishes. +/// +/// So the lock is per bucket, not per config file: a per-file key would let +/// exactly that pair run concurrently. +/// +/// Callers that hold this guard must use [`update_under_transaction_lock`] +/// rather than [`update`] — see that function. +pub async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result { + acquire_transaction_lock_with_sys(&get_bucket_metadata_sys()?, bucket).await +} + +async fn acquire_transaction_lock_with_sys( + sys: &Arc>, + bucket: &str, +) -> Result { + // Resolve the store under a short-lived read guard: this runs before the + // write guard in `acquire_config_write_guards`, and must not still hold a + // read guard when the namespace lock is awaited. + let api = sys.read().await.object_store(); let lock = api - .new_ns_lock(RUSTFS_META_BUCKET, &bucket_targets_transaction_lock_key(bucket)) + .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket)) .await?; Ok(lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?) } -fn bucket_targets_transaction_lock_key(bucket: &str) -> String { +/// The lock resource name is deliberately still the `bucket-targets` one it +/// had when only replication-target writes took it. The key is what nodes +/// agree on, so renaming it would leave a mixed-version cluster with two +/// disjoint keys — and old and new nodes would stop excluding each other on +/// the very writes that are serialized today. +fn bucket_metadata_transaction_lock_key(bucket: &str) -> String { format!("bucket-targets/{bucket}/transaction.lock") } -pub async fn delete(bucket: &str, config_file: &str) -> Result { - let bucket_meta_sys_lock = get_bucket_metadata_sys()?; - let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; - - bucket_meta_sys.delete(bucket, config_file).await -} - pub async fn get_bucket_policy(bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let bucket_meta_sys = bucket_meta_sys_lock.read().await; @@ -864,11 +929,11 @@ impl BucketMetadataSys { } async fn update_and_parse(&mut self, bucket: &str, config_file: &str, data: Vec, parse: bool) -> Result { - let Some(store) = runtime_sources::object_store_handle() else { - return Err(Error::other("errServerNotInitialized")); - }; - - let mut bm = Self::load_bucket_metadata_for_update(store, bucket, parse).await?; + // Load through this system's own store, the one `save` persists to + // (backlog#1052 S7). Reading from the ambient handle instead made the + // read and the write of a single read-modify-write able to target + // different instances. + let mut bm = Self::load_bucket_metadata_for_update(self.api.clone(), bucket, parse).await?; let updated = bm.update_config(config_file, data)?; @@ -1815,6 +1880,160 @@ mod tests { ); } + /// Two metadata systems over one backing store: the in-process stand-in + /// for two nodes. They share no `RwLock`, so nothing but the transaction + /// lock can serialize them — exactly the cross-node case. + async fn two_nodes_over_one_store() -> (Vec, Arc>, Arc>) + { + let (dirs, ecstore) = isolated_store_over_temp_disks().await; + let node_a = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore.clone()))); + let node_b = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore))); + (dirs, node_a, node_b) + } + + /// Writers on different nodes updating *different* config files of one + /// bucket must both survive. Each rewrites the whole metadata blob, so + /// without a lock spanning the read-modify-write the later save carries + /// the earlier writer's field back to its pre-update value — losing an + /// orthogonal config while both clients were told the write succeeded. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn concurrent_config_writes_from_separate_nodes_do_not_lose_writes() { + use crate::bucket::metadata::{BUCKET_POLICY_CONFIG, BUCKET_TAGGING_CONFIG}; + + let (_dirs, node_a, node_b) = two_nodes_over_one_store().await; + let bucket = "cross-node-config-writes"; + node_a + .read() + .await + .persist_and_set(BucketMetadata::new(bucket)) + .await + .expect("initial metadata should persist"); + + // Several rounds: a single pass can serialize by luck, but a lost + // update only needs one interleaving to show up. + const ROUNDS: usize = 8; + for round in 0..ROUNDS { + let tagging = format!("{round}").into_bytes(); + let policy = format!(r#"{{"Version":"2012-10-17","Round":{round}}}"#).into_bytes(); + + let start = Arc::new(tokio::sync::Barrier::new(2)); + let tagging_writer = { + let (node, start, tagging) = (node_a.clone(), start.clone(), tagging.clone()); + tokio::spawn(async move { + start.wait().await; + update_with_sys(node, bucket, BUCKET_TAGGING_CONFIG, tagging).await + }) + }; + let policy_writer = { + let (node, start, policy) = (node_b.clone(), start.clone(), policy.clone()); + tokio::spawn(async move { + start.wait().await; + update_with_sys(node, bucket, BUCKET_POLICY_CONFIG, policy).await + }) + }; + + tagging_writer + .await + .expect("tagging writer should join") + .expect("tagging update should succeed"); + policy_writer + .await + .expect("policy writer should join") + .expect("policy update should succeed"); + + // Disk truth, not either node's cache: the losing write is the one + // that never reached the metadata file. + let persisted = node_a + .read() + .await + .get_config_from_disk(bucket) + .await + .expect("metadata should load from disk"); + assert_eq!( + persisted.tagging_config_xml, tagging, + "round {round}: the policy write clobbered the concurrent tagging write" + ); + assert_eq!( + persisted.policy_config_json, policy, + "round {round}: the tagging write clobbered the concurrent policy write" + ); + } + } + + /// The guard has to cover the load as well as the save. If it were taken + /// only around the save, a second node could load between the two and + /// still overwrite with pre-update state. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn bucket_metadata_transaction_lock_blocks_a_concurrent_config_write() { + use crate::bucket::metadata::BUCKET_TAGGING_CONFIG; + + let (_dirs, node_a, node_b) = two_nodes_over_one_store().await; + let bucket = "cross-node-transaction-lock"; + node_a + .read() + .await + .persist_and_set(BucketMetadata::new(bucket)) + .await + .expect("initial metadata should persist"); + + let held = acquire_transaction_lock_with_sys(&node_a, bucket) + .await + .expect("transaction lock should be acquirable"); + + let blocked = tokio::spawn({ + let node_b = node_b.clone(); + async move { update_with_sys(node_b, bucket, BUCKET_TAGGING_CONFIG, b"".to_vec()).await } + }); + + // Long enough for the write to have finished had it not waited: the + // whole read-modify-write against temp disks is far quicker than this. + tokio::time::sleep(Duration::from_millis(500)).await; + assert!( + !blocked.is_finished(), + "a config write must not proceed while another node holds the bucket's transaction lock" + ); + + drop(held); + + let updated = timeout(Duration::from_secs(10), blocked) + .await + .expect("the blocked write should proceed once the lock is released") + .expect("blocked writer should join"); + updated.expect("the write should succeed after acquiring the lock"); + } + + /// A holder of the transaction lock must not call the locking entry + /// point: the namespace lock is not reentrant, so it would block on + /// itself until the acquire timeout. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn transaction_lock_is_not_reentrant() { + let (_dirs, node_a, node_b) = two_nodes_over_one_store().await; + let bucket = "transaction-lock-reentrancy"; + + let held = acquire_transaction_lock_with_sys(&node_a, bucket) + .await + .expect("first acquisition should succeed"); + + // Same store, hence the same locker owner: exclusion must not depend + // on the two acquisitions coming from different owners. Either + // outcome is acceptable — still waiting, or refused — as long as no + // second guard is handed out. + let reacquired = timeout(Duration::from_millis(500), acquire_transaction_lock_with_sys(&node_b, bucket)).await; + assert!( + !matches!(reacquired, Ok(Ok(_))), + "the transaction lock must exclude a second holder even under the same owner" + ); + + drop(held); + timeout(Duration::from_secs(10), acquire_transaction_lock_with_sys(&node_b, bucket)) + .await + .expect("re-acquisition should not time out once released") + .expect("the lock should be acquirable after release"); + } + fn target(bucket: &str, id: &str) -> BucketTarget { BucketTarget { source_bucket: bucket.to_string(), diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index c94ca6f75..223cb484d 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -7205,7 +7205,7 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st return bucket_status; }; let _targets_guard = lock_bucket_targets_metadata(bucket).await; - let _transaction_guard = match metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await { + let _transaction_guard = match metadata_sys::acquire_bucket_metadata_transaction_lock(bucket).await { Ok(guard) => guard, Err(_) => { bucket_status.status = "failed".to_string(); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 352c42f7d..98e4ea9aa 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -2478,7 +2478,7 @@ async fn start_replication_resync(bucket: &str, reset: &ReplicationResetStartReq }; let _targets_guard = lock_bucket_targets_metadata(bucket).await; - let _transaction_guard = metadata_sys::acquire_bucket_targets_transaction_lock(bucket) + let _transaction_guard = metadata_sys::acquire_bucket_metadata_transaction_lock(bucket) .await .map_err(ApiError::from)?; let (config, _) = metadata_sys::get_replication_config(bucket).await.map_err(ApiError::from)?; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index f5fb811e7..b8a3924b9 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -282,8 +282,8 @@ pub(crate) mod metadata_sys { crate::storage::storage_api::update_bucket_metadata_config(bucket, config_file, data).await } - pub(crate) async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { - crate::storage::storage_api::acquire_bucket_targets_transaction_lock(bucket).await + pub(crate) async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result { + crate::storage::storage_api::acquire_bucket_metadata_transaction_lock(bucket).await } pub(crate) async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 296636264..7d93ac4bf 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -834,7 +834,7 @@ pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String]) - }; for bucket in buckets { - let _transaction_guard = ecstore_bucket::metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await?; + let _transaction_guard = ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock(bucket).await?; let status = pool.get_bucket_resync_status(bucket).await?; if status.targets_map.is_empty() { continue; @@ -1458,8 +1458,8 @@ pub(crate) async fn update_bucket_metadata_config( Ok(updated_at) } -pub(crate) async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { - ecstore_bucket::metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await +pub(crate) async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result { + ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock(bucket).await } pub(crate) async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result {