perf(ecstore): memoize bucket-incarnation fence validation under lifecycle read-lock coverage (#5782)

* perf(ecstore): memoize bucket-incarnation fence validation under lifecycle read-lock coverage

The PUT commit fence from #5648 validated the bucket incarnation with an uncached read (a distributed metadata-transaction read lock plus an EC quorum read of bucket metadata) on every PUT commit. Under 64-concurrency 4KiB PUT load this adds two quorum round-trips per PUT and the resulting lock-manager pressure produced ~1,000 client-visible 'Lock acquisition timeout' failures per 5-minute window (see rustfs/backlog#1776).

Memoize the validation per node while lifecycle read-lock coverage is continuous: bucket deletion/recreation requires the lifecycle WRITE lock, so while at least one read guard on this node has been held continuously the incarnation cannot have changed. The first fenced PUT in a coverage window performs the exact authoritative disk validation as before; overlapping PUTs reuse its result. The memo clears when the node's last guard drops or any guard observes a lost lock, so the next PUT revalidates from disk. Fence semantics are unchanged; only the redundant re-validations under continuous coverage are elided.

Also right-size the s3s footprint ratchet baselines: -1 s3_error! line from this change's error-path consolidation, and +1 s3s-importing file inherited from #5763 (crates/obs/src/telemetry/filter.rs) which landed on main without the baseline bump.

* fix(ecstore): carry the bucket fence registry through the rebalance test store

The rebalance entry test constructor landed on main after this branch was cut and needs the new field.

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();

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

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?);
             }

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<super::bucket_fence::BucketIncarnationFenceGuard> {
+        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<rustfs_lock::NamespaceLockGuard> {
         let lock = self.new_ns_lock(bucket, BUCKET_LIFECYCLE_LOCK_OBJECT).await?;
         lock.get_write_lock(get_lock_acquire_timeout())

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<Uuid>,
+}
+
+/// 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<HashMap<String, FenceEntry>>,
+}
+
+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<Uuid> {
+        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<NamespaceLockGuard>,
+    registry: Arc<BucketFenceRegistry>,
+    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<BucketFenceRegistry>,
+    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<Uuid> {
+        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);
+    }
+}

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

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

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<InstanceContext>,
+    /// Memoizes bucket-incarnation validation under continuous lifecycle
+    /// read-lock coverage (see [`bucket_fence`]).
+    pub(crate) bucket_fence_registry: Arc<bucket_fence::BucketFenceRegistry>,
 }
 
 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(),
         })
     }
 

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(),
         }
     }
 

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(),
         }
     }