diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 6f5a9aec7..954c6e0b0 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -55,12 +55,7 @@ pub async fn init_bucket_metadata_sys(api: Arc, buckets: Vec) { let sys = Arc::new(RwLock::new(sys)); - // Same fail-fast as the old process-global `.set().unwrap()`, scoped to - // the owning instance: double-initializing one store's metadata is a bug. - assert!( - instance_ctx.init_bucket_metadata_sys(sys.clone()), - "bucket metadata sys should be initialized once per instance" - ); + instance_ctx.init_bucket_metadata_sys(sys.clone()); if is_dist_erasure { start_refresh_buckets_metadata_loop(sys); diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 737627a3f..beb7aba25 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -146,7 +146,12 @@ pub struct InstanceContext { /// instance's store; holds the store handle plus the bucket→metadata /// cache, so it is inherently per-store. Replaces the process-global /// bucket-metadata static (whose double-init panicked process-wide). - bucket_metadata_sys: OnceLock>>, + /// + /// Uses `std::sync::Mutex` instead of `OnceLock` so test fixtures that + /// share the bootstrap context can reinitialize the metadata system with + /// their own `ECStore` without panicking. Production startup still sets + /// this exactly once. + bucket_metadata_sys: std::sync::Mutex>>>, /// This instance's background-services cancellation token (Phase 5 Slice 13, /// backlog#939). /// @@ -186,7 +191,7 @@ impl InstanceContext { local_disk_map: Arc::new(RwLock::new(HashMap::new())), local_disk_id_map: Arc::new(RwLock::new(HashMap::new())), local_disk_set_drives: Arc::new(RwLock::new(Vec::new())), - bucket_metadata_sys: OnceLock::new(), + bucket_metadata_sys: std::sync::Mutex::new(None), background_cancel_token: OnceLock::new(), } } @@ -319,16 +324,23 @@ impl InstanceContext { self.local_disk_set_drives.clone() } - /// Set this instance's bucket metadata system (once). Returns `false` if - /// it was already initialized — the caller preserves the old fail-fast - /// contract, now scoped to the instance instead of the process. + /// Set this instance's bucket metadata system. Replaces any existing + /// value so test fixtures that share the bootstrap context can + /// reinitialize with their own store. Returns `true` if this is the + /// first initialization (no previous value existed). pub fn init_bucket_metadata_sys(&self, sys: Arc>) -> bool { - self.bucket_metadata_sys.set(sys).is_ok() + let mut guard = self.bucket_metadata_sys.lock().expect("bucket_metadata_sys lock poisoned"); + let is_first = guard.is_none(); + *guard = Some(sys); + is_first } /// This instance's bucket metadata system, if initialized. pub fn bucket_metadata_sys(&self) -> Option>> { - self.bucket_metadata_sys.get().cloned() + self.bucket_metadata_sys + .lock() + .expect("bucket_metadata_sys lock poisoned") + .clone() } /// Set this instance's background-services cancellation token (once). @@ -387,7 +399,10 @@ impl std::fmt::Debug for InstanceContext { .field("expiry_state_set", &self.expiry_state.get().is_some()) .field("transition_state_set", &self.transition_state.get().is_some()) .field("bucket_monitor_set", &self.bucket_monitor.get().is_some()) - .field("bucket_metadata_sys_set", &self.bucket_metadata_sys.get().is_some()) + .field( + "bucket_metadata_sys_set", + &self.bucket_metadata_sys.lock().map(|g| g.is_some()).unwrap_or(false), + ) .field("replication_stats_set", &self.replication_stats.get().is_some()) .field("replication_pool_set", &self.replication_pool.get().is_some()) .field("background_cancel_token_set", &self.background_cancel_token.get().is_some()) diff --git a/rustfs/src/app/delete_objects_stat_gating_test.rs b/rustfs/src/app/delete_objects_stat_gating_test.rs index 6f3135518..f68776969 100644 --- a/rustfs/src/app/delete_objects_stat_gating_test.rs +++ b/rustfs/src/app/delete_objects_stat_gating_test.rs @@ -26,82 +26,14 @@ //! report per-key results, versioned batch deletes still create delete //! markers and preserve the underlying version. -use super::storage_api::test::bucket::metadata_sys; -use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions, MakeBucketOptions}; +use super::gating_test_env::shared_gating_ecstore; +use super::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions}; use super::storage_api::test::contract::object::{ObjectIO as _, ObjectOperations as _}; -use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; use super::storage_api::test::{StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader}; use crate::storage::storage_api::{StorageObjectLockDeleteOptions, StorageObjectToDelete as ObjectToDelete}; use serial_test::serial; -use std::path::PathBuf; -use std::sync::{Arc, OnceLock}; -use tempfile::TempDir; -use tokio::fs; -use tokio_util::sync::CancellationToken; use uuid::Uuid; -static DELETE_GATING_ENV: OnceLock<(Vec, Arc, TempDir)> = OnceLock::new(); - -async fn setup_delete_gating_env() -> Arc { - if let Some((_paths, store, _)) = DELETE_GATING_ENV.get() { - return store.clone(); - } - - let temp_dir = TempDir::new().expect("create temp dir for delete gating test"); - let temp_path = temp_dir.path().to_path_buf(); - - let disk_paths = vec![ - temp_path.join("disk1"), - temp_path.join("disk2"), - temp_path.join("disk3"), - temp_path.join("disk4"), - ]; - for disk_path in &disk_paths { - fs::create_dir_all(disk_path).await.unwrap(); - } - - let mut endpoints = Vec::new(); - for (i, disk_path) in disk_paths.iter().enumerate() { - let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); - endpoint.set_pool_index(0); - endpoint.set_set_index(0); - endpoint.set_disk_index(i); - endpoints.push(endpoint); - } - - let pool_endpoints = PoolEndpoints { - legacy: false, - set_count: 1, - drives_per_set: 4, - endpoints: Endpoints::from(endpoints), - cmd_line: "delete-objects-stat-gating-test".to_string(), - platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), - }; - - let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); - super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone()) - .await - .unwrap(); - - let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); - let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) - .await - .unwrap(); - - let buckets_list = ecstore - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .unwrap(); - let buckets = buckets_list.into_iter().map(|v| v.name).collect(); - metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; - - let _ = DELETE_GATING_ENV.set((disk_paths, ecstore.clone(), temp_dir)); - ecstore -} - fn compliance_retention_metadata() -> std::collections::HashMap { let retain_until = time::OffsetDateTime::now_utc() + time::Duration::days(30); let mut user_defined = std::collections::HashMap::new(); @@ -118,7 +50,7 @@ fn compliance_retention_metadata() -> std::collections::HashMap #[tokio::test] #[serial] async fn object_lock_bucket_batch_delete_keeps_held_lock_protection() { - let ecstore = setup_delete_gating_env().await; + let ecstore = shared_gating_ecstore().await; let bucket = format!("hp8-lock-{}", Uuid::new_v4()); ecstore @@ -188,7 +120,7 @@ async fn object_lock_bucket_batch_delete_keeps_held_lock_protection() { #[tokio::test] #[serial] async fn non_lock_versioned_bucket_batch_delete_still_creates_delete_marker() { - let ecstore = setup_delete_gating_env().await; + let ecstore = shared_gating_ecstore().await; let bucket = format!("hp8-versioned-{}", Uuid::new_v4()); ecstore @@ -263,7 +195,7 @@ async fn non_lock_versioned_bucket_batch_delete_still_creates_delete_marker() { #[tokio::test] #[serial] async fn non_lock_unversioned_bucket_batch_delete_reports_per_key_results() { - let ecstore = setup_delete_gating_env().await; + let ecstore = shared_gating_ecstore().await; let bucket = format!("hp8-plain-{}", Uuid::new_v4()); ecstore diff --git a/rustfs/src/app/gating_test_env.rs b/rustfs/src/app/gating_test_env.rs new file mode 100644 index 000000000..32587ca09 --- /dev/null +++ b/rustfs/src/app/gating_test_env.rs @@ -0,0 +1,99 @@ +// 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. + +//! Shared test fixture for gating tests (delete_objects_stat_gating_test, +//! put_prelookup_gating_test). +//! +//! Both test modules exercise bucket-metadata-dependent logic against a real +//! 4-disk `ECStore`. Because `ECStore::new()` adopts the process-level +//! bootstrap `InstanceContext` (a single `OnceLock`), calling +//! `init_bucket_metadata_sys` from two separate test modules would panic on +//! the second call. This module provides a single shared setup so only one +//! `ECStore` and one metadata-sys initialization exist per test binary. + +use super::storage_api::test::bucket::metadata_sys; +use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions}; +use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; +use std::path::PathBuf; +use std::sync::{Arc, OnceLock}; +use tempfile::TempDir; +use tokio::fs; +use tokio_util::sync::CancellationToken; + +static SHARED_GATING_ENV: OnceLock<(Vec, Arc, TempDir)> = OnceLock::new(); + +/// Return a shared 4-disk `ECStore` with bucket metadata initialized. +/// +/// The first caller creates the store and initializes the metadata system; +/// subsequent callers get the same `Arc`. Safe to call from +/// `#[serial]` tests that share the process bootstrap context. +pub(crate) async fn shared_gating_ecstore() -> Arc { + if let Some((_paths, store, _)) = SHARED_GATING_ENV.get() { + return store.clone(); + } + + let temp_dir = TempDir::new().expect("create temp dir for gating test env"); + let temp_path = temp_dir.path().to_path_buf(); + + let disk_paths = vec![ + temp_path.join("disk1"), + temp_path.join("disk2"), + temp_path.join("disk3"), + temp_path.join("disk4"), + ]; + for disk_path in &disk_paths { + fs::create_dir_all(disk_path).await.unwrap(); + } + + let mut endpoints = Vec::new(); + for (i, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + endpoints.push(endpoint); + } + + let pool_endpoints = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "gating-test-env".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }; + + let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); + super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone()) + .await + .unwrap(); + + let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) + .await + .unwrap(); + + let buckets_list = ecstore + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .unwrap(); + let buckets = buckets_list.into_iter().map(|v| v.name).collect(); + metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; + + let _ = SHARED_GATING_ENV.set((disk_paths, ecstore.clone(), temp_dir)); + ecstore +} diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs index ca73f337f..0c0c5cffe 100644 --- a/rustfs/src/app/mod.rs +++ b/rustfs/src/app/mod.rs @@ -31,6 +31,8 @@ mod capacity_dirty_scope_test; #[cfg(test)] mod delete_objects_stat_gating_test; #[cfg(test)] +mod gating_test_env; +#[cfg(test)] mod lifecycle_transition_api_test; #[cfg(test)] mod put_prelookup_gating_test; diff --git a/rustfs/src/app/put_prelookup_gating_test.rs b/rustfs/src/app/put_prelookup_gating_test.rs index 754f8d469..743fa1a27 100644 --- a/rustfs/src/app/put_prelookup_gating_test.rs +++ b/rustfs/src/app/put_prelookup_gating_test.rs @@ -23,84 +23,16 @@ //! - an unknown bucket (metadata lookup error) fails closed (gate = true), //! so a degraded metadata subsystem can never silently drop the WORM check. +use super::gating_test_env::shared_gating_ecstore; use super::object_usecase::put_prelookup_worm_gate; -use super::storage_api::test::bucket::metadata_sys; -use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions, MakeBucketOptions}; -use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; +use super::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions}; use serial_test::serial; -use std::path::PathBuf; -use std::sync::{Arc, OnceLock}; -use tempfile::TempDir; -use tokio::fs; -use tokio_util::sync::CancellationToken; use uuid::Uuid; -static PUT_GATING_ENV: OnceLock<(Vec, Arc, TempDir)> = OnceLock::new(); - -async fn setup_put_gating_env() -> Arc { - if let Some((_paths, store, _)) = PUT_GATING_ENV.get() { - return store.clone(); - } - - let temp_dir = TempDir::new().expect("create temp dir for put prelookup gating test"); - let temp_path = temp_dir.path().to_path_buf(); - - let disk_paths = vec![ - temp_path.join("disk1"), - temp_path.join("disk2"), - temp_path.join("disk3"), - temp_path.join("disk4"), - ]; - for disk_path in &disk_paths { - fs::create_dir_all(disk_path).await.unwrap(); - } - - let mut endpoints = Vec::new(); - for (i, disk_path) in disk_paths.iter().enumerate() { - let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); - endpoint.set_pool_index(0); - endpoint.set_set_index(0); - endpoint.set_disk_index(i); - endpoints.push(endpoint); - } - - let pool_endpoints = PoolEndpoints { - legacy: false, - set_count: 1, - drives_per_set: 4, - endpoints: Endpoints::from(endpoints), - cmd_line: "put-prelookup-gating-test".to_string(), - platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), - }; - - let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); - super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone()) - .await - .unwrap(); - - let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); - let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) - .await - .unwrap(); - - let buckets_list = ecstore - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .unwrap(); - let buckets = buckets_list.into_iter().map(|v| v.name).collect(); - metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; - - let _ = PUT_GATING_ENV.set((disk_paths, ecstore.clone(), temp_dir)); - ecstore -} - #[tokio::test] #[serial] async fn worm_gate_keeps_prelookup_for_object_lock_bucket() { - let ecstore = setup_put_gating_env().await; + let ecstore = shared_gating_ecstore().await; let bucket = format!("put-gate-lock-{}", Uuid::new_v4()); ecstore @@ -123,7 +55,7 @@ async fn worm_gate_keeps_prelookup_for_object_lock_bucket() { #[tokio::test] #[serial] async fn worm_gate_allows_skip_for_plain_bucket() { - let ecstore = setup_put_gating_env().await; + let ecstore = shared_gating_ecstore().await; let bucket = format!("put-gate-plain-{}", Uuid::new_v4()); ecstore @@ -140,7 +72,7 @@ async fn worm_gate_allows_skip_for_plain_bucket() { #[tokio::test] #[serial] async fn worm_gate_fails_closed_when_bucket_metadata_is_unavailable() { - let _ecstore = setup_put_gating_env().await; + let _ecstore = shared_gating_ecstore().await; let missing_bucket = format!("put-gate-missing-{}", Uuid::new_v4()); assert!(