fix(ecstore): batch concurrent fresh authority reads

crates/ecstore/src/bucket/artifact.rs

@@ -18,8 +18,10 @@
 mod backup_retention;
 mod delete;
 pub mod inventory;
+mod ordinary_classification;
 #[doc(hidden)]
 pub use delete::ArtifactDeleteTarget;
+pub(crate) use ordinary_classification::OrdinaryClassificationRegistry;
 
 use super::versioning::VersioningApi as _;
 use super::{metadata::BucketMetadata, metadata_sys};
@@ -437,10 +439,16 @@ impl ECStore {
         }
         // Receiver classification must not migrate a legacy incarnation or
         // reacquire the bucket namespace lock held by a DeleteBucket caller.
-        let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket)
-            .await
-            .map_err(|_| DiskError::FileAccessDenied)?;
-        if !persisted || metadata.artifact_base_prefix.is_some() {
+        let allowed = self
+            .ordinary_classification
+            .classify(bucket, async {
+                match metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await {
+                    Ok((metadata, persisted)) => persisted && metadata.artifact_base_prefix.is_none(),
+                    Err(_) => false,
+                }
+            })
+            .await;
+        if !allowed {
             return Err(DiskError::FileAccessDenied);
         }
         Ok(())

crates/ecstore/src/bucket/artifact/ordinary_classification.rs

@@ -0,0 +1,287 @@
+// 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.
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+
+/// Shares only reads started after a caller arrives; retains no idle results.
+#[derive(Default)]
+pub(crate) struct OrdinaryClassificationRegistry {
+    gates: Mutex<HashMap<String, Weak<Gate>>>,
+    #[cfg(test)]
+    after_unregister: Option<std::sync::Barrier>,
+}
+
+#[derive(Default)]
+struct Gate {
+    started_generation: AtomicU64,
+    completed: tokio::sync::Mutex<Option<CompletedRead>>,
+}
+
+struct CompletedRead {
+    generation: u64,
+    allowed: bool,
+}
+
+struct Registration<'a> {
+    registry: &'a OrdinaryClassificationRegistry,
+    bucket: &'a str,
+    gate: Option<Arc<Gate>>,
+    observed_generation: u64,
+}
+
+impl OrdinaryClassificationRegistry {
+    pub(crate) async fn classify(&self, bucket: &str, read: impl Future<Output = bool>) -> bool {
+        self.register(bucket).classify(read).await
+    }
+
+    fn register<'a>(&'a self, bucket: &'a str) -> Registration<'a> {
+        // Callers requiring physical admission must register after acquiring it.
+        // Release the registry lock before awaiting the gate, then acquire the
+        // existing metadata locks.
+        let mut gates = self.gates.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        let gate = gates.get(bucket).and_then(Weak::upgrade).unwrap_or_else(|| {
+            let gate = Arc::new(Gate::default());
+            gates.insert(bucket.to_owned(), Arc::downgrade(&gate));
+            gate
+        });
+        let observed_generation = gate.started_generation.load(Ordering::Acquire);
+        Registration {
+            registry: self,
+            bucket,
+            gate: Some(gate),
+            observed_generation,
+        }
+    }
+}
+
+impl Registration<'_> {
+    async fn classify(self, read: impl Future<Output = bool>) -> bool {
+        let Some(gate) = self.gate.as_ref() else {
+            return false;
+        };
+        let mut completed = gate.completed.lock().await;
+        // Only a read started after this caller registered can authorize it.
+        // Arrivals during a read therefore wait for the next fresh read.
+        if let Some(result) = completed
+            .as_ref()
+            .filter(|result| result.generation > self.observed_generation)
+        {
+            return result.allowed;
+        }
+        let Ok(previous) = gate
+            .started_generation
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |generation| generation.checked_add(1))
+        else {
+            return false;
+        };
+        // Keep the gate through the original bounded metadata read. Cancellation
+        // releases it without publishing a result; another caller can read again.
+        let allowed = read.await;
+        *completed = Some(CompletedRead {
+            generation: previous + 1,
+            allowed,
+        });
+        allowed
+    }
+}
+
+impl Drop for Registration<'_> {
+    fn drop(&mut self) {
+        let Some(gate) = self.gate.take() else {
+            return;
+        };
+        let mut gates = self.registry.gates.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        // Strong references are acquired and released under this registry lock,
+        // so concurrent final callers cannot both miss the last reference.
+        if Arc::strong_count(&gate) == 1
+            && gates
+                .get(self.bucket)
+                .is_some_and(|registered| registered.ptr_eq(&Arc::downgrade(&gate)))
+        {
+            gates.remove(self.bucket);
+        }
+        drop(gate);
+        drop(gates);
+        #[cfg(test)]
+        if let Some(barrier) = &self.registry.after_unregister {
+            barrier.wait();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::poll;
+    use std::task::Poll;
+    use tokio::sync::oneshot;
+
+    fn assert_empty(registry: &OrdinaryClassificationRegistry) {
+        assert!(registry.gates.lock().expect("classification registry").is_empty());
+    }
+
+    #[tokio::test]
+    async fn late_arrivals_share_only_the_next_fail_closed_read() {
+        let registry = OrdinaryClassificationRegistry::default();
+        let reads = AtomicU64::new(0);
+        let (release, waiting) = oneshot::channel();
+        let mut first = Box::pin(registry.classify("bucket", async {
+            reads.fetch_add(1, Ordering::Relaxed);
+            waiting.await.expect("release first read");
+            true
+        }));
+        assert!(poll!(first.as_mut()).is_pending());
+        let mut followers = (0..16)
+            .map(|_| {
+                Box::pin(registry.classify("bucket", async {
+                    reads.fetch_add(1, Ordering::Relaxed);
+                    false
+                }))
+            })
+            .collect::<Vec<_>>();
+        for follower in &mut followers {
+            assert!(poll!(follower.as_mut()).is_pending());
+        }
+        release.send(()).expect("first reader still alive");
+        assert_eq!(poll!(first.as_mut()), Poll::Ready(true));
+        for follower in &mut followers {
+            assert_eq!(poll!(follower.as_mut()), Poll::Ready(false));
+        }
+        assert_eq!(reads.load(Ordering::Relaxed), 2);
+        assert_empty(&registry);
+    }
+
+    #[tokio::test]
+    async fn successive_waves_always_read_again() {
+        let registry = OrdinaryClassificationRegistry::default();
+        let reads = AtomicU64::new(0);
+        for expected in [true, false, true] {
+            assert_eq!(
+                registry
+                    .classify("bucket", async {
+                        reads.fetch_add(1, Ordering::Relaxed);
+                        expected
+                    })
+                    .await,
+                expected
+            );
+            assert_empty(&registry);
+        }
+        assert_eq!(reads.load(Ordering::Relaxed), 3);
+    }
+
+    #[tokio::test]
+    async fn canceled_leader_does_not_publish_or_strand_waiters() {
+        let registry = OrdinaryClassificationRegistry::default();
+        let reads = AtomicU64::new(0);
+        let old_registration = registry.register("bucket");
+        assert!(
+            registry
+                .classify("bucket", async {
+                    reads.fetch_add(1, Ordering::Relaxed);
+                    true
+                })
+                .await
+        );
+        let mut leader = Box::pin(registry.classify("bucket", async {
+            reads.fetch_add(1, Ordering::Relaxed);
+            std::future::pending::<bool>().await
+        }));
+        assert!(poll!(leader.as_mut()).is_pending());
+        let mut old_waiter =
+            Box::pin(old_registration.classify(async { panic!("old waiter must reuse the completed post-registration read") }));
+        assert!(poll!(old_waiter.as_mut()).is_pending());
+        let mut late_waiter = Box::pin(registry.classify("bucket", async {
+            reads.fetch_add(1, Ordering::Relaxed);
+            false
+        }));
+        assert!(poll!(late_waiter.as_mut()).is_pending());
+        drop(leader);
+        assert_eq!(poll!(old_waiter.as_mut()), Poll::Ready(true));
+        assert_eq!(poll!(late_waiter.as_mut()), Poll::Ready(false));
+        assert_eq!(reads.load(Ordering::Relaxed), 3);
+        assert_empty(&registry);
+    }
+
+    #[tokio::test]
+    async fn canceled_follower_does_not_retain_the_gate() {
+        let registry = OrdinaryClassificationRegistry::default();
+        let (release, waiting) = oneshot::channel();
+        let mut leader = Box::pin(registry.classify("bucket", async {
+            waiting.await.expect("release leader");
+            true
+        }));
+        assert!(poll!(leader.as_mut()).is_pending());
+        let mut follower = Box::pin(registry.classify("bucket", async { panic!("canceled follower must not read") }));
+        assert!(poll!(follower.as_mut()).is_pending());
+        drop(follower);
+        assert_eq!(registry.gates.lock().expect("classification registry").len(), 1);
+        release.send(()).expect("leader still alive");
+        assert_eq!(poll!(leader.as_mut()), Poll::Ready(true));
+        assert_empty(&registry);
+    }
+
+    #[test]
+    fn concurrent_final_registrations_remove_the_gate() {
+        let registry = OrdinaryClassificationRegistry {
+            after_unregister: Some(std::sync::Barrier::new(2)),
+            ..Default::default()
+        };
+        let first = registry.register("bucket");
+        let second = registry.register("bucket");
+        // Both destructors reach the unlocked barrier before either returns.
+        // Moving the Arc release after this barrier leaves a stale registry key.
+        std::thread::scope(|scope| {
+            scope.spawn(move || drop(first));
+            scope.spawn(move || drop(second));
+        });
+        assert_empty(&registry);
+    }
+
+    #[tokio::test]
+    async fn buckets_and_stores_do_not_share_gates() {
+        let first_store = OrdinaryClassificationRegistry::default();
+        let second_store = OrdinaryClassificationRegistry::default();
+        let mut blocked = Box::pin(first_store.classify("first", std::future::pending()));
+        assert!(poll!(blocked.as_mut()).is_pending());
+        assert_eq!(poll!(Box::pin(first_store.classify("second", async { true }))), Poll::Ready(true));
+        assert_eq!(poll!(Box::pin(second_store.classify("first", async { false }))), Poll::Ready(false));
+        assert_empty(&second_store);
+        drop(blocked);
+        assert_empty(&first_store);
+    }
+
+    #[tokio::test]
+    async fn generation_exhaustion_fails_closed_without_reading() {
+        let registry = OrdinaryClassificationRegistry::default();
+        let registration = registry.register("bucket");
+        let gate = registration.gate.as_ref().expect("registered gate");
+        gate.started_generation.store(u64::MAX, Ordering::Release);
+        *gate.completed.lock().await = Some(CompletedRead {
+            generation: u64::MAX,
+            allowed: true,
+        });
+        assert!(
+            !registry
+                .classify("bucket", async { panic!("exhausted generation must deny without reading") })
+                .await
+        );
+        assert_eq!(gate.started_generation.load(Ordering::Acquire), u64::MAX);
+        drop(registration);
+        assert_empty(&registry);
+    }
+}

crates/ecstore/src/core/pools.rs

@@ -29378,6 +29378,7 @@ mod pools_tests {
             pool_meta_save_gate: tokio::sync::Mutex::new(super::PoolMetaWriteState::for_test_bootstrap()),
             ctx,
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         })
     }
 

crates/ecstore/src/core/pools_test.rs

@@ -499,6 +499,7 @@ mod decommission_lock_order_tests {
             ),
             ctx,
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         });
         crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
         (store, refresh_calls)

crates/ecstore/src/data_usage/mod.rs

@@ -3334,6 +3334,7 @@ mod tests {
             pool_meta_save_gate: TokioMutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         })
     }
 

crates/ecstore/src/services/rebalance/mod.rs

@@ -88,6 +88,7 @@ pub async fn test_store_with_persisted_rebalance_meta(
         pool_meta_save_gate: tokio::sync::Mutex::default(),
         ctx,
         bucket_fence_registry: std::sync::Arc::default(),
+        ordinary_classification: Default::default(),
     });
     (temp_dirs, store)
 }
@@ -266,6 +267,7 @@ async fn test_pool_stores_with_contexts(
             pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()),
             ctx: store_ctx,
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         })
     };
     let store = make_store(ctx);

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -3415,6 +3415,7 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc<crate::store::ECSt
         pool_meta_save_gate: tokio::sync::Mutex::default(),
         ctx: crate::runtime::instance::bootstrap_ctx(),
         bucket_fence_registry: std::sync::Arc::default(),
+        ordinary_classification: Default::default(),
     })
 }
 

crates/ecstore/src/set_disk/ops/object.rs

@@ -13071,6 +13071,7 @@ pub(crate) mod hermetic_set_disks_support {
             pool_meta_save_gate: tokio::sync::Mutex::default(),
             ctx: Arc::clone(&instance_ctx),
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         });
         let shutdown = tokio_util::sync::CancellationToken::new();
         instance_ctx

crates/ecstore/src/store/heal.rs

@@ -2082,6 +2082,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         }
     }
 
@@ -4185,6 +4186,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         };
 
         let err = store

crates/ecstore/src/store/init.rs

@@ -620,6 +620,7 @@ impl ECStore {
             // this point) and later reads share one cell.
             ctx: instance_ctx.clone(),
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         });
 
         // Only set it when this instance's deployment ID is not yet configured

crates/ecstore/src/store/mod.rs

@@ -494,6 +494,7 @@ pub struct ECStore {
     /// Memoizes bucket-incarnation validation under continuous lifecycle
     /// read-lock coverage (see [`bucket_fence`]).
     pub(crate) bucket_fence_registry: Arc<bucket_fence::BucketFenceRegistry>,
+    pub(crate) ordinary_classification: crate::bucket::artifact::OrdinaryClassificationRegistry,
 }
 
 const METRIC_SCANNER_DATA_MOVEMENT_PAUSED: &str = "rustfs_scanner_data_movement_paused";
@@ -1904,6 +1905,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         };
 
         let rendered = format!("{store:?}");
@@ -1992,6 +1994,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
+            ordinary_classification: Default::default(),
         })
     }
 

crates/ecstore/src/store/multipart.rs

@@ -1663,6 +1663,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         }
     }
 

crates/ecstore/src/store/object.rs

@@ -9762,6 +9762,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         }
     }
 
@@ -9871,6 +9872,7 @@ mod tests {
             pool_meta_save_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: std::sync::Arc::default(),
+            ordinary_classification: Default::default(),
         }
     }