gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-31 02:22:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): serialize every bucket config write under the transaction lock (#5445)

Browse Source 
Only replication-target writes took the bucket transaction lock. Every other
config write (policy, tagging, lifecycle, versioning, ...) went straight to
the process-local metadata-system guard, which serializes nothing across
nodes.

Each config write is a read-modify-write of one whole BucketMetadata blob:
load the blob, replace one field, save the blob 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 the same
bucket therefore 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.

Route update(), delete() and update_config_with() through
acquire_config_write_guards(), which takes the cluster-wide transaction lock
first and the metadata-system write guard second. That order is load-bearing:
taking 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.

The lock is per bucket rather than per config file, since a per-file key
would let exactly the offending pair run concurrently. Rename the helper to
acquire_bucket_metadata_transaction_lock to match, but deliberately keep the
lock resource string as "bucket-targets/{bucket}/transaction.lock": the key
is what nodes agree on, so renaming it would leave a mixed-version cluster
with two disjoint keys and stop old and new nodes from excluding each other
on the very writes that are serialized today.

update_config_with() already narrowed its staleness window to a single load
and save, but its exclusion was explicitly process-local; it is now
cluster-wide, so its doc comment no longer disclaims cross-node races.

Also make update_and_parse load through self.api instead of the ambient store
handle, so the read and the write of one read-modify-write cannot resolve to
different instances.

The new tests drive two BucketMetadataSys instances over one ECStore -- the
in-process stand-in for two nodes, since they share no RwLock and can only be
serialized by the namespace lock. Verified the lost-update test has teeth by
removing the lock and confirming it fails on round 0, with the tagging config
clobbered to empty by the concurrent policy write.
This commit is contained in:
Zhengchao An
2026-07-30 01:17:09 +08:00
committed by GitHub
parent 2dea4a9acf
commit 88fa3877c1
6 changed files with 273 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/api/mod.rs
+2 -2
View File
@@ -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,
};
}
crates/ecstore/src/bucket/metadata_sys.rs
+264 -45
View File
@@ -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<u8>) -> Result<OffsetDateTime> {
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<OffsetDateTime> {
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<RwLock<BucketMetadataSys>>,
bucket: &str,
config_file: &str,
data: Vec<u8>,
) -> Result<OffsetDateTime> {
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<RwLock<BucketMetadataSys>>, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
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<RwLock<BucketMetadataSys>>,
bucket: &str,
) -> Result<(rustfs_lock::NamespaceLockGuard, tokio::sync::OwnedRwLockWriteGuard<BucketMetadataSys>)> {
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<u8>) -> Result<OffsetDateTime> {
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<u8>) -> Result<OffsetDateTime> {
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<F>(bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
where
F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + 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<rustfs_lock::NamespaceLockGuard> {
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<rustfs_lock::NamespaceLockGuard> {
acquire_transaction_lock_with_sys(&get_bucket_metadata_sys()?, bucket).await
}
async fn acquire_transaction_lock_with_sys(
sys: &Arc<RwLock<BucketMetadataSys>>,
bucket: &str,
) -> Result<rustfs_lock::NamespaceLockGuard> {
// 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<OffsetDateTime> {
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<u8>, parse: bool) -> Result<OffsetDateTime> {
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<tempfile::TempDir>, Arc<RwLock<BucketMetadataSys>>, Arc<RwLock<BucketMetadataSys>>)
{
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!("<Tagging><Round>{round}</Round></Tagging>").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"<Tagging/>".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(),
rustfs/src/admin/handlers/site_replication.rs
+1 -1
View File
@@ -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();
rustfs/src/admin/router.rs
+1 -1
View File
@@ -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)?;
rustfs/src/admin/storage_api.rs
+2 -2
View File
@@ -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<rustfs_lock::NamespaceLockGuard> {
crate::storage::storage_api::acquire_bucket_targets_transaction_lock(bucket).await
pub(crate) async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result<rustfs_lock::NamespaceLockGuard> {
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<u8>) -> Result<OffsetDateTime> {
rustfs/src/storage/storage_api.rs
+3 -3
View File
@@ -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<rustfs_lock::NamespaceLockGuard> {
ecstore_bucket::metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await
pub(crate) async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result<rustfs_lock::NamespaceLockGuard> {
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<u8>) -> Result<time::OffsetDateTime> {