mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 16:48:58 +00:00
refactor(ecstore): move the bucket metadata system into InstanceContext (#4622)
backlog#1052 S3, third slice — the hard blocker for a second embedded server's service stage. GLOBAL_BUCKET_METADATA_SYS was a process-global OnceLock whose init ran .set().unwrap(): a second instance's service startup panicked the whole process. The system it guards is inherently per-store (it holds the store handle and that store's bucket→metadata cache), so it now lives on the store's own InstanceContext: - init_bucket_metadata_sys writes the cell of the store it was handed (api.ctx) — no signature change — and keeps the double-init fail-fast, now scoped to the instance (assert on the per-context set). The dist-erasure check for the refresh loop reads the same context. - get_global_bucket_metadata_sys / the crate-private getter behind the ~20 get_*_config helpers resolve through the current_ctx() facade — the published store's context, or bootstrap before that — so every existing reader keeps its exact single-instance behavior. - New acceptance test: two real stores in one process each initialize their own metadata system (the old global cell panicked right there), plus a shared builder extracted from the store-graph test. 201 bucket/metadata regression tests green.
This commit is contained in:
@@ -23,7 +23,6 @@ use crate::runtime::sources as runtime_sources;
|
|||||||
use crate::storage_api_contracts::heal::HealOperations as _;
|
use crate::storage_api_contracts::heal::HealOperations as _;
|
||||||
use crate::store::ECStore;
|
use crate::store::ECStore;
|
||||||
use futures::future::join_all;
|
use futures::future::join_all;
|
||||||
use lazy_static::lazy_static;
|
|
||||||
use rustfs_common::heal_channel::HealOpts;
|
use rustfs_common::heal_channel::HealOpts;
|
||||||
use rustfs_policy::policy::BucketPolicy;
|
use rustfs_policy::policy::BucketPolicy;
|
||||||
use s3s::dto::ReplicationConfiguration;
|
use s3s::dto::ReplicationConfiguration;
|
||||||
@@ -33,7 +32,6 @@ use s3s::dto::{
|
|||||||
Tagging, VersioningConfiguration, WebsiteConfiguration,
|
Tagging, VersioningConfiguration, WebsiteConfiguration,
|
||||||
};
|
};
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::sync::OnceLock;
|
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use std::{collections::HashMap, sync::Arc};
|
use std::{collections::HashMap, sync::Arc};
|
||||||
use time::OffsetDateTime;
|
use time::OffsetDateTime;
|
||||||
@@ -42,35 +40,44 @@ use tokio::time::sleep;
|
|||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
use tracing::{error, warn};
|
use tracing::{error, warn};
|
||||||
|
|
||||||
lazy_static! {
|
|
||||||
pub static ref GLOBAL_BUCKET_METADATA_SYS: OnceLock<Arc<RwLock<BucketMetadataSys>>> = OnceLock::new();
|
|
||||||
}
|
|
||||||
|
|
||||||
const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
|
const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
|
||||||
|
|
||||||
pub async fn init_bucket_metadata_sys(api: Arc<ECStore>, buckets: Vec<String>) {
|
pub async fn init_bucket_metadata_sys(api: Arc<ECStore>, buckets: Vec<String>) {
|
||||||
|
// The metadata system is inherently per-store (it holds the store handle
|
||||||
|
// and that store's bucket cache), so it lives on the store's own instance
|
||||||
|
// context (backlog#1052 S3) — a second instance initializes its own cell
|
||||||
|
// instead of panicking on the process-global one.
|
||||||
|
let instance_ctx = api.ctx.clone();
|
||||||
|
let is_dist_erasure = instance_ctx.is_dist_erasure().await;
|
||||||
|
|
||||||
let mut sys = BucketMetadataSys::new(api);
|
let mut sys = BucketMetadataSys::new(api);
|
||||||
sys.init(buckets).await;
|
sys.init(buckets).await;
|
||||||
|
|
||||||
let sys = Arc::new(RwLock::new(sys));
|
let sys = Arc::new(RwLock::new(sys));
|
||||||
|
|
||||||
GLOBAL_BUCKET_METADATA_SYS.set(sys.clone()).unwrap();
|
// 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"
|
||||||
|
);
|
||||||
|
|
||||||
if runtime_sources::setup_is_dist_erasure().await {
|
if is_dist_erasure {
|
||||||
start_refresh_buckets_metadata_loop(sys);
|
start_refresh_buckets_metadata_loop(sys);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The current instance's bucket metadata system (legacy free-function
|
||||||
|
/// facade: resolves the published store's context, or the bootstrap one).
|
||||||
pub fn get_global_bucket_metadata_sys() -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
pub fn get_global_bucket_metadata_sys() -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
GLOBAL_BUCKET_METADATA_SYS.get().cloned()
|
crate::runtime::global::current_ctx().bucket_metadata_sys()
|
||||||
}
|
}
|
||||||
|
|
||||||
// panic if not init
|
|
||||||
pub(super) fn get_bucket_metadata_sys() -> Result<Arc<RwLock<BucketMetadataSys>>> {
|
pub(super) fn get_bucket_metadata_sys() -> Result<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
if let Some(sys) = GLOBAL_BUCKET_METADATA_SYS.get() {
|
if let Some(sys) = get_global_bucket_metadata_sys() {
|
||||||
Ok(sys.clone())
|
Ok(sys)
|
||||||
} else {
|
} else {
|
||||||
Err(Error::other("GLOBAL_BUCKET_METADATA_SYS not init"))
|
Err(Error::other("bucket metadata sys not initialized for this instance"))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -42,6 +42,7 @@
|
|||||||
use super::global::TypeLocalDiskSetDrives;
|
use super::global::TypeLocalDiskSetDrives;
|
||||||
use crate::bucket::bandwidth::monitor::Monitor;
|
use crate::bucket::bandwidth::monitor::Monitor;
|
||||||
use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState};
|
use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState};
|
||||||
|
use crate::bucket::metadata_sys::BucketMetadataSys;
|
||||||
use crate::bucket::replication::{DynReplicationPool, ReplicationStats};
|
use crate::bucket::replication::{DynReplicationPool, ReplicationStats};
|
||||||
use crate::disk::DiskStore;
|
use crate::disk::DiskStore;
|
||||||
use crate::layout::endpoints::{EndpointServerPools, SetupType};
|
use crate::layout::endpoints::{EndpointServerPools, SetupType};
|
||||||
@@ -139,6 +140,13 @@ pub struct InstanceContext {
|
|||||||
local_disk_map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
|
local_disk_map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
|
||||||
local_disk_id_map: Arc<RwLock<HashMap<Uuid, String>>>,
|
local_disk_id_map: Arc<RwLock<HashMap<Uuid, String>>>,
|
||||||
local_disk_set_drives: Arc<RwLock<TypeLocalDiskSetDrives>>,
|
local_disk_set_drives: Arc<RwLock<TypeLocalDiskSetDrives>>,
|
||||||
|
/// This instance's bucket metadata system (backlog#1052 S3).
|
||||||
|
///
|
||||||
|
/// Set once when storage startup initializes bucket metadata for this
|
||||||
|
/// 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>>>,
|
||||||
/// This instance's background-services cancellation token (Phase 5 Slice 13,
|
/// This instance's background-services cancellation token (Phase 5 Slice 13,
|
||||||
/// backlog#939).
|
/// backlog#939).
|
||||||
///
|
///
|
||||||
@@ -178,6 +186,7 @@ impl InstanceContext {
|
|||||||
local_disk_map: Arc::new(RwLock::new(HashMap::new())),
|
local_disk_map: Arc::new(RwLock::new(HashMap::new())),
|
||||||
local_disk_id_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())),
|
local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
|
||||||
|
bucket_metadata_sys: OnceLock::new(),
|
||||||
background_cancel_token: OnceLock::new(),
|
background_cancel_token: OnceLock::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -310,6 +319,18 @@ impl InstanceContext {
|
|||||||
self.local_disk_set_drives.clone()
|
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.
|
||||||
|
pub fn init_bucket_metadata_sys(&self, sys: Arc<RwLock<BucketMetadataSys>>) -> bool {
|
||||||
|
self.bucket_metadata_sys.set(sys).is_ok()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// This instance's bucket metadata system, if initialized.
|
||||||
|
pub fn bucket_metadata_sys(&self) -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
|
self.bucket_metadata_sys.get().cloned()
|
||||||
|
}
|
||||||
|
|
||||||
/// Set this instance's background-services cancellation token (once).
|
/// Set this instance's background-services cancellation token (once).
|
||||||
pub fn init_background_cancel_token(&self, token: CancellationToken) -> Result<(), CancellationToken> {
|
pub fn init_background_cancel_token(&self, token: CancellationToken) -> Result<(), CancellationToken> {
|
||||||
self.background_cancel_token.set(token)
|
self.background_cancel_token.set(token)
|
||||||
@@ -366,6 +387,7 @@ impl std::fmt::Debug for InstanceContext {
|
|||||||
.field("expiry_state_set", &self.expiry_state.get().is_some())
|
.field("expiry_state_set", &self.expiry_state.get().is_some())
|
||||||
.field("transition_state_set", &self.transition_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_monitor_set", &self.bucket_monitor.get().is_some())
|
||||||
|
.field("bucket_metadata_sys_set", &self.bucket_metadata_sys.get().is_some())
|
||||||
.field("replication_stats_set", &self.replication_stats.get().is_some())
|
.field("replication_stats_set", &self.replication_stats.get().is_some())
|
||||||
.field("replication_pool_set", &self.replication_pool.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())
|
.field("background_cancel_token_set", &self.background_cancel_token.get().is_some())
|
||||||
|
|||||||
@@ -783,17 +783,13 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Phase 5 follow-up (backlog#1052): building a real store through the
|
// Build a real 4-drive store over a temp dir around a fresh instance
|
||||||
// ctx-explicit constructor lands every construction-time write — object
|
// context — shared by the isolation tests below.
|
||||||
// graph adoption, local-disk registry, deployment id — on the passed
|
async fn build_isolated_test_store(
|
||||||
// context, not on the process bootstrap one. This is the storage-layer
|
temp_dir: &std::path::Path,
|
||||||
// seam a future second embedded server needs to stay isolated.
|
cmd_line: &str,
|
||||||
#[tokio::test]
|
) -> (Arc<crate::runtime::instance::InstanceContext>, Arc<crate::store::ECStore>) {
|
||||||
async fn new_with_instance_ctx_threads_context_through_store_graph() {
|
let disk_paths: Vec<_> = (1..=4).map(|i| temp_dir.join(format!("disk{i}"))).collect();
|
||||||
use crate::runtime::instance::InstanceContext;
|
|
||||||
|
|
||||||
let temp_dir = tempfile::tempdir().expect("create temp store dir");
|
|
||||||
let disk_paths: Vec<_> = (1..=4).map(|i| temp_dir.path().join(format!("disk{i}"))).collect();
|
|
||||||
for path in &disk_paths {
|
for path in &disk_paths {
|
||||||
tokio::fs::create_dir_all(path).await.expect("create disk dir");
|
tokio::fs::create_dir_all(path).await.expect("create disk dir");
|
||||||
}
|
}
|
||||||
@@ -811,11 +807,11 @@ mod tests {
|
|||||||
set_count: 1,
|
set_count: 1,
|
||||||
drives_per_set: 4,
|
drives_per_set: 4,
|
||||||
endpoints: Endpoints::from(endpoints),
|
endpoints: Endpoints::from(endpoints),
|
||||||
cmd_line: "instance-ctx-store-graph-test".to_string(),
|
cmd_line: cmd_line.to_string(),
|
||||||
platform: "test".to_string(),
|
platform: "test".to_string(),
|
||||||
}]);
|
}]);
|
||||||
|
|
||||||
let instance_ctx = Arc::new(InstanceContext::new());
|
let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
|
||||||
crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
|
crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
|
||||||
.await
|
.await
|
||||||
.expect("register local disks into the fresh context");
|
.expect("register local disks into the fresh context");
|
||||||
@@ -829,6 +825,19 @@ mod tests {
|
|||||||
.await
|
.await
|
||||||
.expect("store should build around the fresh context");
|
.expect("store should build around the fresh context");
|
||||||
|
|
||||||
|
(instance_ctx, store)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Phase 5 follow-up (backlog#1052): building a real store through the
|
||||||
|
// ctx-explicit constructor lands every construction-time write — object
|
||||||
|
// graph adoption, local-disk registry, deployment id — on the passed
|
||||||
|
// context, not on the process bootstrap one. This is the storage-layer
|
||||||
|
// seam a future second embedded server needs to stay isolated.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn new_with_instance_ctx_threads_context_through_store_graph() {
|
||||||
|
let temp_dir = tempfile::tempdir().expect("create temp store dir");
|
||||||
|
let (instance_ctx, store) = build_isolated_test_store(temp_dir.path(), "instance-ctx-store-graph-test").await;
|
||||||
|
|
||||||
assert!(
|
assert!(
|
||||||
Arc::ptr_eq(&store.ctx, &instance_ctx),
|
Arc::ptr_eq(&store.ctx, &instance_ctx),
|
||||||
"the store must adopt the explicitly passed instance context"
|
"the store must adopt the explicitly passed instance context"
|
||||||
@@ -862,4 +871,28 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// backlog#1052 S3: two stores in one process each initialize their own
|
||||||
|
// bucket metadata system on their own instance context. Before this, the
|
||||||
|
// second `init_bucket_metadata_sys` panicked on the process-global
|
||||||
|
// OnceLock — the hard blocker for a second embedded server's services.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn two_stores_initialize_their_own_bucket_metadata_sys() {
|
||||||
|
let temp_a = tempfile::tempdir().expect("create temp store dir a");
|
||||||
|
let temp_b = tempfile::tempdir().expect("create temp store dir b");
|
||||||
|
let (ctx_a, store_a) = build_isolated_test_store(temp_a.path(), "bucket-metadata-isolation-a").await;
|
||||||
|
let (ctx_b, store_b) = build_isolated_test_store(temp_b.path(), "bucket-metadata-isolation-b").await;
|
||||||
|
|
||||||
|
crate::bucket::metadata_sys::init_bucket_metadata_sys(store_a.clone(), Vec::new()).await;
|
||||||
|
// The old process-global cell would panic right here.
|
||||||
|
crate::bucket::metadata_sys::init_bucket_metadata_sys(store_b.clone(), Vec::new()).await;
|
||||||
|
|
||||||
|
let sys_a = ctx_a
|
||||||
|
.bucket_metadata_sys()
|
||||||
|
.expect("store A's context must hold its metadata system");
|
||||||
|
let sys_b = ctx_b
|
||||||
|
.bucket_metadata_sys()
|
||||||
|
.expect("store B's context must hold its metadata system");
|
||||||
|
assert!(!Arc::ptr_eq(&sys_a, &sys_b), "each store must own a distinct bucket metadata system");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user