diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index c2b671e6b..51bd51088 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -607,6 +607,7 @@ mod tests { start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), }); let mut version = FileInfo::new("object.bin", 4, 2); version.name = "object.bin".to_string(); diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index 401d39c8f..7dcddda97 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -2454,6 +2454,7 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() { start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), }); let err = store diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c58184062..8f9c4bbc1 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -982,10 +982,9 @@ impl SetDisks { bucket_lifecycle_guard = Some( metadata_sys::object_store_in(&self.ctx) .await? - .acquire_bucket_lifecycle_read_lock(bucket) + .acquire_bucket_incarnation_fence(bucket, expected_incarnation_id) .await?, ); - self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?; } object_lock_guard = Some( self.acquire_write_lock_diag("put_object_precondition", bucket, object) @@ -1321,10 +1320,9 @@ impl SetDisks { bucket_lifecycle_guard = Some( metadata_sys::object_store_in(&self.ctx) .await? - .acquire_bucket_lifecycle_read_lock(bucket) + .acquire_bucket_incarnation_fence(bucket, expected_incarnation_id) .await?, ); - self.validate_bucket_incarnation(bucket, expected_incarnation_id).await?; } object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 6055f8bf8..080a86441 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -176,6 +176,52 @@ impl ECStore { }) } + /// Acquire the bucket lifecycle read lock and validate the bucket + /// incarnation against `expected`, memoizing the validation while this + /// node keeps continuous read-lock coverage (see [`super::bucket_fence`]). + /// + /// Semantics are identical to the pre-existing per-PUT + /// `acquire_bucket_lifecycle_read_lock` + from-disk + /// `validate_bucket_incarnation` pair: the first PUT in a coverage window + /// performs exactly that authoritative disk validation; overlapping PUTs + /// reuse its result, which is sound because bucket deletion/recreation + /// requires the lifecycle WRITE lock and therefore cannot have run while + /// any read guard was continuously held. + pub(crate) async fn acquire_bucket_incarnation_fence( + &self, + bucket: &str, + expected: uuid::Uuid, + ) -> Result { + let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?; + let pieces = super::bucket_fence::FencePieces { + registry: self.bucket_fence_registry.clone(), + inner, + }; + let memoized = pieces.enter(bucket); + let current = match memoized { + Some(current) => current, + None => match metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await { + Ok(current) => { + // Never memoize under lost coverage: a granted lifecycle + // write lock could already have changed the incarnation. + if !pieces.lock_lost() { + pieces.memoize(bucket, current); + } + current + } + Err(err) => { + pieces.abandon(bucket); + return Err(err); + } + }, + }; + if current != expected { + pieces.abandon(bucket); + return Err(StorageError::BucketNotFound(bucket.to_string())); + } + Ok(pieces.into_guard(bucket)) + } + pub(crate) async fn acquire_bucket_lifecycle_write_lock(&self, bucket: &str) -> Result { let lock = self.new_ns_lock(bucket, BUCKET_LIFECYCLE_LOCK_OBJECT).await?; lock.get_write_lock(get_lock_acquire_timeout()) diff --git a/crates/ecstore/src/store/bucket_fence.rs b/crates/ecstore/src/store/bucket_fence.rs new file mode 100644 index 000000000..b7d13a234 --- /dev/null +++ b/crates/ecstore/src/store/bucket_fence.rs @@ -0,0 +1,215 @@ +// 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. + +//! Memoized bucket-incarnation validation under continuous lifecycle read-lock +//! coverage. +//! +//! The PUT commit fence introduced by #5648 validated the bucket incarnation +//! with an uncached read (`get_bucket_incarnation_id_from_disk`: a distributed +//! metadata-transaction read lock plus an EC quorum read of the bucket +//! metadata) on every PUT commit. Under small-object write load that is two +//! extra quorum round-trips per PUT, and the resulting lock-manager pressure +//! produced sustained `Lock acquisition timeout` errors (~1,000 client-visible +//! failures per 5-minute 64-concurrency window in benchmarks). +//! +//! The memo exploits the fence's own locking protocol: bucket deletion and +//! recreation take the bucket lifecycle WRITE lock, while every fenced PUT +//! holds a lifecycle READ lock for the whole commit. Therefore, while at least +//! one lifecycle read guard on this node has been held continuously, no +//! lifecycle write lock can have been granted anywhere in the cluster, so the +//! bucket incarnation cannot have changed. The first fenced PUT in such a +//! coverage window pays the authoritative disk validation exactly as before; +//! subsequent PUTs whose guards overlap that window compare against the +//! memoized value. When the node's last guard drops — or any guard observes +//! `is_lock_lost` — the memo is cleared and the next PUT revalidates from +//! disk. +//! +//! The memo is deliberately per-node process state (not a cross-node cache): +//! its validity is derived purely from locks this process itself holds, so +//! best-effort peer cache invalidation (which is why the fence read from disk +//! in the first place) is irrelevant to its correctness. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use rustfs_lock::NamespaceLockGuard; +use uuid::Uuid; + +#[derive(Default)] +struct FenceEntry { + guards: usize, + validated: Option, +} + +/// Per-store registry tracking, per bucket, how many lifecycle read guards are +/// live on this node and the incarnation id validated under that coverage. +#[derive(Default)] +pub(crate) struct BucketFenceRegistry { + entries: Mutex>, +} + +impl BucketFenceRegistry { + /// Register a new live guard for `bucket` and return the memoized + /// incarnation id if one is valid for the current coverage window. + fn enter(&self, bucket: &str) -> Option { + let mut entries = self.entries.lock().expect("bucket fence registry poisoned"); + let entry = entries.entry(bucket.to_string()).or_default(); + entry.guards += 1; + entry.validated + } + + /// Memoize `incarnation` for `bucket`. Only meaningful while the caller + /// still holds a registered guard (which it does by construction). + fn memoize(&self, bucket: &str, incarnation: Uuid) { + let mut entries = self.entries.lock().expect("bucket fence registry poisoned"); + if let Some(entry) = entries.get_mut(bucket) + && entry.guards > 0 + { + entry.validated = Some(incarnation); + } + } + + /// Deregister a guard. Clears the memo when the last guard leaves or when + /// the leaving guard lost its lock (lost coverage means a lifecycle write + /// lock may have been granted, so the memo can no longer be trusted). + fn exit(&self, bucket: &str, lock_lost: bool) { + let mut entries = self.entries.lock().expect("bucket fence registry poisoned"); + if let Some(entry) = entries.get_mut(bucket) { + entry.guards = entry.guards.saturating_sub(1); + if lock_lost { + entry.validated = None; + } + if entry.guards == 0 { + entries.remove(bucket); + } + } + } +} + +/// A held bucket lifecycle read lock plus its registration in the fence +/// registry. Dropping the guard deregisters it; the memo is cleared when the +/// last guard for the bucket drops (or a lost lock is observed). +pub(crate) struct BucketIncarnationFenceGuard { + inner: Option, + registry: Arc, + bucket: String, +} + +impl BucketIncarnationFenceGuard { + pub(crate) fn is_lock_lost(&self) -> bool { + self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) + } +} + +impl Drop for BucketIncarnationFenceGuard { + fn drop(&mut self) { + let lost = self.is_lock_lost(); + self.registry.exit(&self.bucket, lost); + self.inner.take(); + } +} + +pub(super) struct FencePieces { + pub(super) registry: Arc, + pub(super) inner: NamespaceLockGuard, +} + +impl FencePieces { + /// Register the freshly acquired read lock and return the memoized + /// incarnation for the coverage window, if any. + pub(super) fn enter(&self, bucket: &str) -> Option { + self.registry.enter(bucket) + } + + pub(super) fn memoize(&self, bucket: &str, incarnation: Uuid) { + self.registry.memoize(bucket, incarnation) + } + + pub(super) fn lock_lost(&self) -> bool { + self.inner.is_lock_lost() + } + + pub(super) fn into_guard(self, bucket: &str) -> BucketIncarnationFenceGuard { + BucketIncarnationFenceGuard { + inner: Some(self.inner), + registry: self.registry, + bucket: bucket.to_string(), + } + } + + /// Abandon the acquisition (validation failed): deregister and release. + pub(super) fn abandon(self, bucket: &str) { + let lost = self.lock_lost(); + self.registry.exit(bucket, lost); + drop(self.inner); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn uuid(n: u128) -> Uuid { + Uuid::from_u128(n) + } + + #[test] + fn memo_valid_only_while_guards_overlap() { + let reg = BucketFenceRegistry::default(); + + assert_eq!(reg.enter("b"), None, "first guard sees no memo"); + reg.memoize("b", uuid(1)); + assert_eq!(reg.enter("b"), Some(uuid(1)), "overlapping guard reuses memo"); + reg.exit("b", false); + reg.exit("b", false); + + // Coverage gap: all guards gone, memo must be dropped. + assert_eq!(reg.enter("b"), None, "post-gap guard must revalidate"); + reg.exit("b", false); + } + + #[test] + fn lost_lock_clears_memo_but_keeps_other_guards_registered() { + let reg = BucketFenceRegistry::default(); + + assert_eq!(reg.enter("b"), None); + reg.memoize("b", uuid(7)); + assert_eq!(reg.enter("b"), Some(uuid(7))); + + // First guard exits reporting a lost lock: memo cleared even though + // a second guard is still live. + reg.exit("b", true); + assert_eq!(reg.enter("b"), None, "memo not trusted after a lost lock"); + reg.exit("b", false); + reg.exit("b", false); + } + + #[test] + fn buckets_are_isolated() { + let reg = BucketFenceRegistry::default(); + assert_eq!(reg.enter("a"), None); + reg.memoize("a", uuid(1)); + assert_eq!(reg.enter("b"), None, "memo does not leak across buckets"); + reg.exit("b", false); + reg.exit("a", false); + } + + #[test] + fn memoize_without_live_guard_is_ignored() { + let reg = BucketFenceRegistry::default(); + reg.memoize("b", uuid(9)); + assert_eq!(reg.enter("b"), None); + reg.exit("b", false); + } +} diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 2c90e744e..a77381014 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -284,6 +284,7 @@ mod tests { start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), }; let (result, err) = store diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 3304a4eeb..8680dc767 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -419,6 +419,7 @@ impl ECStore { // legacy path) so startup writes (erasure type recorded before // this point) and later reads share one cell. ctx: instance_ctx.clone(), + bucket_fence_registry: std::sync::Arc::default(), }); // Only set it when this instance's deployment ID is not yet configured diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index cf2bc4c74..d1a7ac850 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -141,6 +141,7 @@ fn should_enqueue_transition_immediately(oi: &ObjectInfo) -> bool { const MAX_UPLOADS_LIST: usize = 10000; mod bucket; +mod bucket_fence; pub(crate) use bucket::await_bucket_namespace_operation; mod heal; mod heal_walk; @@ -193,6 +194,9 @@ pub struct ECStore { /// startup writes and post-construction reads share one cell — single /// instance behavior is unchanged. pub(crate) ctx: Arc, + /// Memoizes bucket-incarnation validation under continuous lifecycle + /// read-lock coverage (see [`bucket_fence`]). + pub(crate) bucket_fence_registry: Arc, } impl std::fmt::Debug for ECStore { @@ -890,6 +894,7 @@ mod tests { start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), ctx, + bucket_fence_registry: Arc::default(), }) } diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index ccf4b5dc1..8b35bf1d7 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -761,6 +761,7 @@ mod tests { start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), } } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 58b37de22..9dbd21369 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -3012,6 +3012,7 @@ mod tests { start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), } } @@ -3052,6 +3053,7 @@ mod tests { start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), } }