gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(test): allow InstanceContext bucket_metadata_sys reinitialization for tests (#4638)

Browse Source 
The OnceLock introduced in 5fbd49a80 for bucket_metadata_sys panics when
multiple test modules initialize their own ECStore against the shared
bootstrap context. Switch to Mutex<Option> so test fixtures can
reinitialize safely.

Extract shared gating test environment to gating_test_env.rs to avoid
duplicate ECStore setup across delete_objects_stat_gating_test and
put_prelookup_gating_test.

Fixes 8 test failures:
- capacity_dirty_scope_test (2 tests)
- delete_objects_stat_gating_test (3 tests)
- put_prelookup_gating_test (3 tests)
This commit is contained in:
Zhengchao An
2026-07-10 04:03:26 +08:00
committed by GitHub
parent 5e80980698
commit cefbe2ccc9
6 changed files with 135 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/metadata_sys.rs
+1 -6
View File
@@ -55,12 +55,7 @@ pub async fn init_bucket_metadata_sys(api: Arc<ECStore>, buckets: Vec<String>) {
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);
crates/ecstore/src/runtime/instance.rs
+23 -8
View File
@@ -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<Arc<RwLock<BucketMetadataSys>>>,
///
/// 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<Option<Arc<RwLock<BucketMetadataSys>>>>,
/// 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<RwLock<BucketMetadataSys>>) -> 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<Arc<RwLock<BucketMetadataSys>>> {
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())
rustfs/src/app/delete_objects_stat_gating_test.rs
+5 -73
View File
@@ -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<PathBuf>, Arc<ECStore>, TempDir)> = OnceLock::new();
async fn setup_delete_gating_env() -> Arc<ECStore> {
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<String, String> {
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<String, String>
#[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
rustfs/src/app/gating_test_env.rs
+99
View File
@@ -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<PathBuf>, Arc<ECStore>, 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<ECStore>`. Safe to call from
/// `#[serial]` tests that share the process bootstrap context.
pub(crate) async fn shared_gating_ecstore() -> Arc<ECStore> {
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
}
rustfs/src/app/mod.rs
+2
View File
@@ -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;
rustfs/src/app/put_prelookup_gating_test.rs
+5 -73
View File
@@ -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<PathBuf>, Arc<ECStore>, TempDir)> = OnceLock::new();
async fn setup_put_gating_env() -> Arc<ECStore> {
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!(