diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 672e290dc..9b7987387 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -326,6 +326,7 @@ pub mod global { } pub mod runtime { + pub use crate::runtime::instance::{InstanceContext, bootstrap_ctx}; pub use crate::runtime::sources::{ boot_time, bucket_monitor, deployment_id, endpoint_pools, expiry_state_handle, first_cluster_node_is_local, global_lock_client, global_lock_clients, global_tier_config_mgr, local_disk_map_read, object_store_handle, region, @@ -389,8 +390,9 @@ pub mod store_list { pub mod storage { pub use crate::store::HealWalkVersion; pub use crate::store::{ - ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_lock_clients, - prewarm_local_disk_id_map, + ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, + init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map, + prewarm_local_disk_id_map_with_instance_ctx, }; } diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 16b6f183b..425f05ec5 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -102,6 +102,21 @@ impl Sets { fm: &FormatV3, pool_idx: usize, parity_count: usize, + ) -> Result> { + Self::new_with_instance_ctx(disks, endpoints, fm, pool_idx, parity_count, bootstrap_ctx()).await + } + + /// Build the pool's sets bound to an explicit instance context (Phase 5 + /// follow-up, backlog#1052). The legacy [`Sets::new`] entry adopts the + /// process bootstrap context; a store constructed around its own context + /// passes it here so the whole object graph shares one cell. + pub async fn new_with_instance_ctx( + disks: Vec>, + endpoints: &PoolEndpoints, + fm: &FormatV3, + pool_idx: usize, + parity_count: usize, + instance_ctx: Arc, ) -> Result> { let set_count = fm.erasure.sets.len(); let set_drive_count = fm.erasure.sets[0].len(); @@ -127,8 +142,8 @@ impl Sets { continue; } - if disk.as_ref().unwrap().is_local() && runtime_sources::setup_is_dist_erasure().await { - let local_disk = runtime_sources::local_disk_set_drive(pool_idx, i, j).await; + if disk.as_ref().unwrap().is_local() && instance_ctx.is_dist_erasure().await { + let local_disk = runtime_sources::local_disk_set_drive(&instance_ctx, pool_idx, i, j).await; if local_disk.is_none() { warn!("sets new set_drive {}-{} local_disk is none", i, j); @@ -163,7 +178,7 @@ impl Sets { .as_ref() .map(|registry| registry.clients_for_endpoints(&set_endpoints)) .unwrap_or_default(); - let set_disks = SetDisks::new( + let set_disks = SetDisks::new_with_instance_ctx( runtime_sources::local_node_name().await, Arc::new(RwLock::new(set_drive)), set_drive_count, @@ -173,6 +188,7 @@ impl Sets { set_endpoints, fm.clone(), lockers, + instance_ctx.clone(), ) .await; @@ -193,10 +209,7 @@ impl Sets { default_parity_count: parity_count, distribution_algo: fm.erasure.distribution_algo.clone(), exit_signal: Some(tx), - // Single-instance: same bootstrap context the owning ECStore adopts - // (constructed before the store, so sourced here directly). Slice 8 - // threads a per-instance context in for true multi-instance. - ctx: bootstrap_ctx(), + ctx: instance_ctx, }); let asets = sets.clone(); @@ -1462,4 +1475,65 @@ mod tests { "unformatted disk must be Missing on its real index, not on a placeholder" ); } + + fn instance_ctx_test_pool_endpoints() -> (FormatV3, PoolEndpoints) { + let format = FormatV3::new(1, 2); + let endpoints = vec![ + Endpoint::try_from("http://127.0.0.1:9000/data0").expect("first endpoint should parse"), + Endpoint::try_from("http://127.0.0.1:9001/data1").expect("second endpoint should parse"), + ]; + let pool_endpoints = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "instance-ctx-adoption-test".to_string(), + platform: "test".to_string(), + }; + (format, pool_endpoints) + } + + // Phase 5 follow-up (backlog#1052): a pool built through the ctx-explicit + // constructor carries the caller's context through Sets AND every SetDisks, + // so nothing in the object graph silently binds to the process bootstrap. + #[tokio::test] + async fn sets_new_with_instance_ctx_threads_context_through_graph() { + let (format, pool_endpoints) = instance_ctx_test_pool_endpoints(); + let instance_ctx = Arc::new(InstanceContext::new()); + + let sets = Sets::new_with_instance_ctx(vec![None, None], &pool_endpoints, &format, 0, 1, instance_ctx.clone()) + .await + .expect("sets should build with empty disks"); + + assert!( + Arc::ptr_eq(sets.instance_ctx(), &instance_ctx), + "Sets must adopt the explicitly passed instance context" + ); + for set_disks in &sets.disk_set { + assert!( + Arc::ptr_eq(set_disks.instance_ctx(), &instance_ctx), + "every SetDisks must adopt the explicitly passed instance context" + ); + } + assert!( + !Arc::ptr_eq(sets.instance_ctx(), &bootstrap_ctx()), + "a fresh context must not alias the process bootstrap context" + ); + } + + // The legacy constructor keeps single-instance behavior byte-for-byte: it + // still adopts the process bootstrap context. + #[tokio::test] + async fn sets_new_legacy_adopts_bootstrap_context() { + let (format, pool_endpoints) = instance_ctx_test_pool_endpoints(); + + let sets = Sets::new(vec![None, None], &pool_endpoints, &format, 0, 1) + .await + .expect("sets should build with empty disks"); + + assert!( + Arc::ptr_eq(sets.instance_ctx(), &bootstrap_ctx()), + "legacy Sets::new must keep adopting the process bootstrap context" + ); + } } diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index b249ad322..94902c6b0 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -20,6 +20,7 @@ use std::{ use crate::bucket::bandwidth::monitor::Monitor; use crate::disk::endpoint::Endpoint; +use crate::runtime::instance::InstanceContext; use crate::{ bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}, bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys}, @@ -33,8 +34,8 @@ use crate::{ TypeLocalDiskSetDrives, get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints, get_global_endpoints_opt, get_global_lock_client, get_global_lock_clients, get_global_region, get_global_tier_config_mgr, global_rustfs_port, init_global_bucket_monitor, is_dist_erasure, is_erasure, is_erasure_sd, - is_first_cluster_node_local, resolve_object_store_handle, set_global_deployment_id, set_global_lock_client, - set_global_lock_clients, set_object_layer, update_erasure_type, + is_first_cluster_node_local, resolve_object_store_handle, set_global_lock_client, set_global_lock_clients, + set_object_layer, update_erasure_type, }, services::batch_processor::{GlobalBatchProcessors, get_global_processors}, services::event_notification::EventNotifier, @@ -278,12 +279,6 @@ pub(crate) fn replication_runtime_initialized() -> bool { crate::runtime::global::current_ctx().replication_initialized() } -pub(crate) fn ensure_deployment_id(deployment_id: Uuid) { - if get_global_deployment_id().is_none() { - set_global_deployment_id(deployment_id); - } -} - pub fn global_lock_client() -> Option> { get_global_lock_client() } @@ -419,8 +414,8 @@ pub(crate) async fn clear_local_disk_id_map_for_test() { local_disk_id_map_handle().write().await.clear(); } -pub(crate) async fn record_local_disk_id(disk_id: Uuid, endpoint: String) { - local_disk_id_map_handle().write().await.insert(disk_id, endpoint); +pub(crate) async fn record_local_disk_id(instance_ctx: &Arc, disk_id: Uuid, endpoint: String) { + instance_ctx.local_disk_id_map().write().await.insert(disk_id, endpoint); } pub(crate) async fn replace_local_disk_id(previous: Option, current: Option, endpoint: String) { @@ -434,8 +429,8 @@ pub(crate) async fn replace_local_disk_id(previous: Option, current: Optio } } -pub(crate) async fn record_local_disks(disks: Vec) { - let map = local_disk_map_handle(); +pub(crate) async fn record_local_disks(instance_ctx: &Arc, disks: Vec) { + let map = instance_ctx.local_disk_map(); let mut global_local_disk_map = map.write().await; for disk in disks { let path = disk.endpoint().to_string(); @@ -443,8 +438,13 @@ pub(crate) async fn record_local_disks(disks: Vec) { } } -pub(crate) async fn local_disk_set_drive(pool_idx: usize, set_idx: usize, disk_idx: usize) -> Option { - local_disk_set_drives_handle().read().await[pool_idx][set_idx][disk_idx].clone() +pub(crate) async fn local_disk_set_drive( + instance_ctx: &Arc, + pool_idx: usize, + set_idx: usize, + disk_idx: usize, +) -> Option { + instance_ctx.local_disk_set_drives().read().await[pool_idx][set_idx][disk_idx].clone() } pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option { @@ -488,8 +488,12 @@ pub(crate) async fn local_disk_entries() -> Vec> { local_disk_map_handle().read().await.values().cloned().collect() } -pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPools, opt: &DiskOption) -> Result<()> { - let set_drives = local_disk_set_drives_handle(); +pub(crate) async fn initialize_local_disk_maps( + instance_ctx: &Arc, + endpoint_pools: EndpointServerPools, + opt: &DiskOption, +) -> Result<()> { + let set_drives = instance_ctx.local_disk_set_drives(); let mut global_set_drives = set_drives.write().await; for pool_eps in endpoint_pools.as_ref().iter() { let mut set_count_drives = Vec::with_capacity(pool_eps.set_count); @@ -500,7 +504,7 @@ pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPoo global_set_drives.push(set_count_drives); } - let map = local_disk_map_handle(); + let map = instance_ctx.local_disk_map(); let mut global_local_disk_map = map.write().await; for pool_eps in endpoint_pools.as_ref().iter() { diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 32011ad74..2e2daa570 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -1744,10 +1744,39 @@ impl SetDisks { format: FormatV3, lockers: Vec>, ) -> Arc { - // Single-instance sources the process bootstrap context (the one the - // owning ECStore adopts). Slice 8 threads a per-instance context in for - // true multi-instance. - let ctx = bootstrap_ctx(); + Self::new_with_instance_ctx( + locker_owner, + disks, + set_drive_count, + default_parity_count, + set_index, + pool_index, + set_endpoints, + format, + lockers, + bootstrap_ctx(), + ) + .await + } + + /// Build a set bound to an explicit instance context (Phase 5 follow-up, + /// backlog#1052). The legacy [`SetDisks::new`] entry adopts the process + /// bootstrap context; a store constructed around its own context threads it + /// down here so the whole object graph shares one cell. + #[allow(clippy::too_many_arguments)] + pub async fn new_with_instance_ctx( + locker_owner: String, + disks: Arc>>>, + set_drive_count: usize, + default_parity_count: usize, + set_index: usize, + pool_index: usize, + set_endpoints: Vec, + format: FormatV3, + lockers: Vec>, + instance_ctx: Arc, + ) -> Arc { + let ctx = instance_ctx; Arc::new(SetDisks { locker_owner, disks, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index cd8925f34..69ec8f479 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -15,6 +15,7 @@ use super::*; use crate::core::pools::local_decommission_queue_prefix; use crate::error::is_err_decommission_running; +use crate::runtime::instance::InstanceContext; use crate::runtime::sources as runtime_sources; use crate::storage_api_contracts::object::EcstoreObjectIO; use tracing::{debug, error, info, warn}; @@ -165,6 +166,24 @@ impl ECStore { #[allow(clippy::new_ret_no_self)] #[instrument(level = "debug", skip(endpoint_pools))] pub async fn new(address: SocketAddr, endpoint_pools: EndpointServerPools, ctx: CancellationToken) -> Result> { + Self::new_with_instance_ctx(address, endpoint_pools, ctx, crate::runtime::instance::bootstrap_ctx()).await + } + + /// Build a store around an explicit instance context (Phase 5 follow-up, + /// backlog#1052). The legacy [`ECStore::new`] entry adopts the process + /// bootstrap context, keeping single-instance startup byte-for-byte + /// unchanged; a caller that owns its own context (a future second embedded + /// server) passes it here so every construction-time write — pool sets, + /// local-disk registry, deployment id — lands on that context instead of + /// the shared bootstrap one. + #[allow(clippy::new_ret_no_self)] + #[instrument(level = "debug", skip(endpoint_pools, instance_ctx))] + pub async fn new_with_instance_ctx( + address: SocketAddr, + endpoint_pools: EndpointServerPools, + ctx: CancellationToken, + instance_ctx: Arc, + ) -> Result> { // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; let mut deployment_id = None; @@ -308,15 +327,16 @@ impl ECStore { } } - let sets = Sets::new(disks.clone(), pool_eps, &fm, i, common_parity_drives).await?; + let sets = + Sets::new_with_instance_ctx(disks.clone(), pool_eps, &fm, i, common_parity_drives, instance_ctx.clone()).await?; pools.push(sets); disk_map.insert(i, disks); } // Replace the local disk - if !runtime_sources::setup_is_dist_erasure().await { - runtime_sources::record_local_disks(local_disks).await; + if !instance_ctx.is_dist_erasure().await { + runtime_sources::record_local_disks(&instance_ctx, local_disks).await; } let peer_sys = S3PeerSys::new(&endpoint_pools); @@ -334,14 +354,17 @@ impl ECStore { decommission_cancelers, start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::new(()), - // Adopt the process bootstrap context so startup writes (erasure - // type recorded before this point) and later reads share one cell. - ctx: crate::runtime::instance::bootstrap_ctx(), + // Adopt the caller's context (the process bootstrap one on the + // legacy path) so startup writes (erasure type recorded before + // this point) and later reads share one cell. + ctx: instance_ctx.clone(), }); - // Only set it when the global deployment ID is not yet configured - if let Some(dep_id) = deployment_id { - runtime_sources::ensure_deployment_id(dep_id); + // Only set it when this instance's deployment ID is not yet configured + if let Some(dep_id) = deployment_id + && instance_ctx.deployment_id().is_none() + { + instance_ctx.set_deployment_id(dep_id); } let wait_sec = 5; @@ -759,4 +782,84 @@ mod tests { "the expanded pool should be initialized by its own first local endpoint" ); } + + // 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() { + 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 { + tokio::fs::create_dir_all(path).await.expect("create disk dir"); + } + + let mut endpoints = Vec::new(); + for (i, path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(path.to_str().expect("disk path should be utf-8")).expect("local endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + endpoints.push(endpoint); + } + let endpoint_pools = EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "instance-ctx-store-graph-test".to_string(), + platform: "test".to_string(), + }]); + + let instance_ctx = Arc::new(InstanceContext::new()); + crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) + .await + .expect("register local disks into the fresh context"); + + let store = crate::store::ECStore::new_with_instance_ctx( + "127.0.0.1:0".parse().expect("test address"), + endpoint_pools, + CancellationToken::new(), + instance_ctx.clone(), + ) + .await + .expect("store should build around the fresh context"); + + assert!( + Arc::ptr_eq(&store.ctx, &instance_ctx), + "the store must adopt the explicitly passed instance context" + ); + for sets in &store.pools { + assert!( + Arc::ptr_eq(sets.instance_ctx(), &instance_ctx), + "every pool's Sets must carry the passed instance context" + ); + } + assert_eq!( + instance_ctx.deployment_id(), + Some(store.id), + "the deployment id must land on the passed context and mirror the store id" + ); + + let registered: Vec = instance_ctx.local_disk_map().read().await.keys().cloned().collect(); + assert_eq!(registered.len(), 4, "the passed context must register all four local disks"); + let bootstrap = crate::runtime::instance::bootstrap_ctx(); + assert_ne!( + bootstrap.deployment_id(), + Some(store.id), + "the bootstrap context must not absorb the fresh store's deployment id" + ); + let bootstrap_map = bootstrap.local_disk_map(); + let bootstrap_map = bootstrap_map.read().await; + for key in ®istered { + assert!( + !bootstrap_map.contains_key(key), + "the bootstrap context must not absorb the fresh store's disks" + ); + } + } } diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 3fa2b0dd2..f80859133 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -164,8 +164,9 @@ pub(crate) mod utils; use peer::init_local_peer; pub use peer::{ - all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks, init_lock_clients, - prewarm_local_disk_id_map, + all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks, + init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map, + prewarm_local_disk_id_map_with_instance_ctx, }; pub struct ECStore { diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 21441efea..ac1062bff 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::runtime::instance::InstanceContext; use crate::runtime::sources as runtime_sources; use tracing::{debug, error}; @@ -22,8 +23,12 @@ const EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed"; async fn remember_local_disk_id(disk: &DiskStore) -> Option { + remember_local_disk_id_with_instance_ctx(&crate::runtime::global::current_ctx(), disk).await +} + +async fn remember_local_disk_id_with_instance_ctx(instance_ctx: &Arc, disk: &DiskStore) -> Option { let disk_id = disk.get_disk_id().await.ok().flatten()?; - runtime_sources::record_local_disk_id(disk_id, disk.endpoint().to_string()).await; + runtime_sources::record_local_disk_id(instance_ctx, disk_id, disk.endpoint().to_string()).await; Some(disk_id) } @@ -69,7 +74,21 @@ pub async fn all_local_disk() -> Vec { } pub async fn prewarm_local_disk_id_map() { - for disk in all_local_disk().await { + prewarm_local_disk_id_map_with_instance_ctx(&crate::runtime::global::current_ctx()).await +} + +/// Prewarm the disk-id map of an explicit instance context (Phase 5 follow-up, +/// backlog#1052): startup passes the context whose disk map it just populated +/// instead of resolving the process-level default. +pub async fn prewarm_local_disk_id_map_with_instance_ctx(instance_ctx: &Arc) { + let disks: Vec = instance_ctx + .local_disk_map() + .read() + .await + .values() + .filter_map(|v| v.as_ref().cloned()) + .collect(); + for disk in disks { if let Err(err) = disk.get_disk_id().await { debug!( event = EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED, @@ -82,17 +101,28 @@ pub async fn prewarm_local_disk_id_map() { continue; } - let _ = remember_local_disk_id(&disk).await; + let _ = remember_local_disk_id_with_instance_ctx(instance_ctx, &disk).await; } } pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> { + init_local_disks_with_instance_ctx(&crate::runtime::global::current_ctx(), endpoint_pools).await +} + +/// Register the pools' local disks into an explicit instance context (Phase 5 +/// follow-up, backlog#1052). The legacy [`init_local_disks`] entry resolves the +/// process-level default context; startup paths that own a context pass it here +/// so a future second instance's disks cannot leak into the first one's registry. +pub async fn init_local_disks_with_instance_ctx( + instance_ctx: &Arc, + endpoint_pools: EndpointServerPools, +) -> Result<()> { let opt = &DiskOption { cleanup: true, health_check: true, }; - runtime_sources::initialize_local_disk_maps(endpoint_pools, opt).await + runtime_sources::initialize_local_disk_maps(instance_ctx, endpoint_pools, opt).await } pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { @@ -182,3 +212,63 @@ pub async fn get_disk_infos(disks: &[Option]) -> Vec res } + +#[cfg(test)] +mod tests { + use super::*; + use crate::layout::endpoints::{Endpoints, PoolEndpoints}; + + fn single_local_disk_pools(dir: &std::path::Path) -> EndpointServerPools { + let mut endpoint = Endpoint::try_from(dir.to_str().expect("temp dir path should be utf-8")).expect("local endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + + EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 1, + endpoints: Endpoints::from(vec![endpoint]), + cmd_line: "instance-ctx-disk-registry-test".to_string(), + platform: "test".to_string(), + }]) + } + + // Phase 5 follow-up (backlog#1052): registering local disks through the + // ctx-explicit entry writes the passed context's registry only — the + // process bootstrap context (and any other instance) stays clean, so a + // future second server's disks cannot leak into the first one's registry. + #[tokio::test] + async fn init_local_disks_with_instance_ctx_isolates_disk_registry() { + let temp_dir = tempfile::tempdir().expect("create temp disk dir"); + let endpoint_pools = single_local_disk_pools(temp_dir.path()); + let instance_ctx = Arc::new(InstanceContext::new()); + + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools) + .await + .expect("local disks should register into the passed context"); + + let registered: Vec = instance_ctx.local_disk_map().read().await.keys().cloned().collect(); + assert_eq!(registered.len(), 1, "the passed context must hold exactly the one local disk"); + assert_eq!( + instance_ctx.local_disk_set_drives().read().await.len(), + 1, + "the passed context must hold the pool/set/drive layout" + ); + + let bootstrap = crate::runtime::instance::bootstrap_ctx(); + let bootstrap_map = bootstrap.local_disk_map(); + let bootstrap_map = bootstrap_map.read().await; + let sibling = InstanceContext::new(); + for key in ®istered { + assert!( + !bootstrap_map.contains_key(key), + "bootstrap context must not absorb a disk registered into an explicit context" + ); + assert!( + !sibling.local_disk_map().read().await.contains_key(key), + "a sibling context must not observe another instance's disks" + ); + } + } +} diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index bb2599c9b..fff5a6803 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -44,7 +44,8 @@ //! //! Only **one `RustFSServer`** may exist per process because the underlying //! storage engine uses process-global singletons (`OnceLock`). Attempting to -//! start a second server will return an error. +//! start a second server will return an error. Removing this limitation is +//! tracked in [backlog#1052](https://github.com/rustfs/backlog/issues/1052). use crate::server::ShutdownHandle; use crate::startup_embedded::{EmbeddedStartedServer, EmbeddedStartupArgs, EmbeddedStartupError, run_embedded_startup}; diff --git a/rustfs/src/startup_embedded.rs b/rustfs/src/startup_embedded.rs index 20bac96c6..0a34481d0 100644 --- a/rustfs/src/startup_embedded.rs +++ b/rustfs/src/startup_embedded.rs @@ -24,6 +24,7 @@ use crate::{ startup_services::init_embedded_startup_runtime_services, startup_shutdown::signal_embedded_startup_shutdown, startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime}, + storage_api::startup::storage::bootstrap_instance_ctx, }; use std::{io, net::SocketAddr, path::PathBuf}; use tokio_util::sync::CancellationToken; @@ -108,6 +109,12 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result Result Result Result<()> { #[instrument(skip(config))] async fn run(config: Config) -> Result<()> { + // Single-instance startup threads the process bootstrap context through + // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future + // multi-instance server constructs its own context here instead. + let instance_ctx = bootstrap_instance_ctx(); + let StartupListenContext { readiness, server_addr, server_address, - } = init_startup_listen_context(&config).await?; + } = init_startup_listen_context(&config, &instance_ctx).await?; - let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes).await?; + let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes, &instance_ctx).await?; let StartupHttpServers { state_manager, s3_shutdown_tx, @@ -119,7 +125,7 @@ async fn run(config: Config) -> Result<()> { let StartupStorageRuntime { store, shutdown_token: ctx, - } = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?; + } = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone(), instance_ctx).await?; let service_runtime = init_startup_runtime_services( &config, diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index f66647509..869dd1e75 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -45,14 +45,17 @@ const EVENT_EMBEDDED_SERVER_STATE: &str = "embedded_server_state"; /// /// Phase 5 (backlog#939) moved per-instance runtime state (erasure setup, /// region, deployment id, endpoints, service handles, disk registry, cancel -/// token) into `ECStore`'s `InstanceContext`, so two `ECStore` object graphs -/// can now stay isolated. This guard is intentionally **retained**: the startup -/// path still publishes into the process-level *bootstrap* context (write-once -/// region/endpoints/deployment id) and the single `GLOBAL_OBJECT_API` handle, -/// so a second startup would fail-fast on that shared state. Lifting the guard -/// requires threading a per-instance context through storage startup (so each -/// server constructs its own context instead of sharing the bootstrap); until -/// then, rejecting the second start is safer than the panic it would become. +/// token) into `ECStore`'s `InstanceContext`, and backlog#1052 S1 threads an +/// explicit context through the storage startup path, so a second server's +/// *storage-layer* state could now stay isolated. This guard is still +/// intentionally **retained**: embedded startup shares the bootstrap context +/// (see `run_embedded_startup`), the request path resolves the store per +/// request through the process-level `GLOBAL_OBJECT_API`/`AppContext` +/// singletons (a second listener would serve the first instance's data), and +/// IAM/bucket-metadata/config/credentials remain process singletons. Lifting +/// the guard is staged in backlog#1052 (S2 per-server dispatch, S3 app +/// subsystems, S4 node-identity globals, then S5 removes this guard); until +/// then, rejecting the second start is safer than the failure it would become. static EMBEDDED_SERVER_STARTED: AtomicBool = AtomicBool::new(false); #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs index 121ea785a..9b9a19beb 100644 --- a/rustfs/src/startup_runtime_sources.rs +++ b/rustfs/src/startup_runtime_sources.rs @@ -16,7 +16,7 @@ use crate::config::RustFSBufferConfig; use crate::runtime_sources::{ current_outbound_tls_generation as runtime_current_outbound_tls_generation, current_replication_pool_handle, }; -use crate::storage_api::startup::runtime_sources::{DynReplicationPool, set_global_region, set_global_rustfs_port}; +use crate::storage_api::startup::runtime_sources::{DynReplicationPool, InstanceContext, set_global_rustfs_port}; use rustfs_kms::KmsServiceManager; use rustfs_obs::{GlobalError as ObservabilityError, OtelGuard}; use rustfs_tls_runtime::{OutboundTlsMaterial, TlsGeneration}; @@ -30,8 +30,8 @@ pub(crate) fn init_action_credentials( rustfs_credentials::init_global_action_credentials(Some(access_key), Some(secret_key)) } -pub(crate) fn publish_region(region: s3s::region::Region) { - set_global_region(region); +pub(crate) fn publish_region(instance_ctx: &Arc, region: s3s::region::Region) { + instance_ctx.set_region(region); } pub(crate) fn publish_server_port(port: u16) { diff --git a/rustfs/src/startup_server.rs b/rustfs/src/startup_server.rs index 439babda8..4fe60fbac 100644 --- a/rustfs/src/startup_server.rs +++ b/rustfs/src/startup_server.rs @@ -17,6 +17,7 @@ use crate::{ config::Config, server::{ServiceState, ServiceStateManager, ShutdownHandle, start_http_server}, startup_runtime_sources, + storage_api::startup::runtime_sources::InstanceContext, }; use rustfs_common::GlobalReadiness; use rustfs_utils::net::parse_and_resolve_address; @@ -76,14 +77,17 @@ pub(crate) struct StartupHttpServers { pub(crate) console_shutdown_tx: Option, } -pub(crate) async fn init_startup_listen_context(config: &Config) -> Result { +pub(crate) async fn init_startup_listen_context( + config: &Config, + instance_ctx: &Arc, +) -> Result { log_sanitized_server_config(config); let readiness = Arc::new(GlobalReadiness::new()); if let Some(region_str) = &config.region { region_str .parse::() - .map(startup_runtime_sources::publish_region) + .map(|region| startup_runtime_sources::publish_region(instance_ctx, region)) .map_err(|err| Error::other(format!("invalid region '{}': {}", region_str, err)))?; } @@ -191,7 +195,10 @@ pub(crate) fn find_embedded_available_port() -> Result { Err(last_err.unwrap_or_else(|| Error::other("failed to reserve an embedded TCP port"))) } -pub(crate) async fn init_embedded_startup_listen_context(config: &Config) -> Result { +pub(crate) async fn init_embedded_startup_listen_context( + config: &Config, + instance_ctx: &Arc, +) -> Result { let readiness = Arc::new(GlobalReadiness::new()); let server_addr = @@ -210,7 +217,7 @@ pub(crate) async fn init_embedded_startup_listen_context(config: &Config) -> Res if let Some(region_str) = &config.region { region_str .parse::() - .map(startup_runtime_sources::publish_region) + .map(|region| startup_runtime_sources::publish_region(instance_ctx, region)) .map_err(|err| Error::other(format!("invalid region '{region_str}': {err}")))?; } diff --git a/rustfs/src/startup_storage.rs b/rustfs/src/startup_storage.rs index 1764789e5..56e090ce4 100644 --- a/rustfs/src/startup_storage.rs +++ b/rustfs/src/startup_storage.rs @@ -14,9 +14,9 @@ use crate::startup_fs_guard::enforce_unsupported_fs_policy; use crate::storage_api::startup::storage::{ - ECStore, EndpointServerPools, global_config_init_error_is_deterministic, init_background_replication, init_ecstore_config, - init_global_config_sys, init_local_disks, init_lock_clients, prewarm_local_disk_id_map, set_global_endpoints, - try_migrate_server_config, update_erasure_type, + ECStore, EndpointServerPools, InstanceContext, global_config_init_error_is_deterministic, init_background_replication, + init_ecstore_config, init_global_config_sys, init_local_disks_with_instance_ctx, init_lock_clients, + prewarm_local_disk_id_map_with_instance_ctx, try_migrate_server_config, }; use rustfs_common::{GlobalReadiness, SystemStage}; use std::{ @@ -44,7 +44,11 @@ pub(crate) struct StartupStorageRuntime { pub(crate) shutdown_token: CancellationToken, } -pub(crate) async fn init_startup_storage_foundation(server_address: &str, volumes: &[String]) -> Result { +pub(crate) async fn init_startup_storage_foundation( + server_address: &str, + volumes: &[String], + instance_ctx: &Arc, +) -> Result { info!( target: "rustfs::main::run", event = EVENT_ENDPOINT_PARSING_STARTED, @@ -71,8 +75,8 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume .map_err(Error::other)?; enforce_unsupported_fs_policy(&endpoint_pools)?; - set_global_endpoints(endpoint_pools.as_ref().clone()); - update_erasure_type(setup_type).await; + instance_ctx.set_endpoints(endpoint_pools.clone()); + instance_ctx.update_erasure_type(setup_type).await; debug!( target: "rustfs::main::run", @@ -83,7 +87,7 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume state = "starting", "starting local disk initialization" ); - init_local_disks(endpoint_pools.clone()) + init_local_disks_with_instance_ctx(instance_ctx, endpoint_pools.clone()) .await .inspect_err(|err| { error!( @@ -98,7 +102,7 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume ); }) .map_err(Error::other)?; - prewarm_local_disk_id_map().await; + prewarm_local_disk_id_map_with_instance_ctx(instance_ctx).await; init_lock_clients(endpoint_pools.clone()); log_storage_pool_layout(&endpoint_pools); @@ -109,16 +113,17 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume pub(crate) async fn init_embedded_startup_storage_foundation( server_address: &str, volumes: &[String], + instance_ctx: &Arc, ) -> Result { let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address, volumes.to_vec()) .await .map_err(|err| Error::other(format!("endpoints: {err}")))?; enforce_unsupported_fs_policy(&endpoint_pools).map_err(|err| Error::other(format!("unsupported fs guard: {err}")))?; - set_global_endpoints(endpoint_pools.as_ref().clone()); - update_erasure_type(setup_type).await; + instance_ctx.set_endpoints(endpoint_pools.clone()); + instance_ctx.update_erasure_type(setup_type).await; - init_local_disks(endpoint_pools.clone()) + init_local_disks_with_instance_ctx(instance_ctx, endpoint_pools.clone()) .await .map_err(|err| Error::other(format!("local disks: {err}")))?; init_lock_clients(endpoint_pools.clone()); @@ -130,6 +135,7 @@ pub(crate) async fn init_startup_storage_runtime( server_addr: SocketAddr, endpoint_pools: &EndpointServerPools, readiness: Arc, + instance_ctx: Arc, ) -> Result { let ctx = CancellationToken::new(); @@ -142,7 +148,7 @@ pub(crate) async fn init_startup_storage_runtime( state = "starting", "starting ECStore initialization" ); - let store = ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()) + let store = ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), ctx.clone(), instance_ctx) .await .inspect_err(|err| { error!( @@ -172,21 +178,23 @@ pub(crate) async fn init_embedded_startup_storage_runtime( endpoint_pools: &EndpointServerPools, readiness: Arc, shutdown_token: CancellationToken, + instance_ctx: Arc, ) -> Result { - let store = match ECStore::new(server_addr, endpoint_pools.clone(), shutdown_token.clone()).await { - Ok(store) => store, - Err(err) => { - error!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = EVENT_EMBEDDED_STORAGE_INIT_FAILED, - stage = "ecstore_new", - error = ?err, - "Embedded storage initialization failed" - ); - return Err(Error::other(format!("ECStore: {err}"))); - } - }; + let store = + match ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), shutdown_token.clone(), instance_ctx).await { + Ok(store) => store, + Err(err) => { + error!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = EVENT_EMBEDDED_STORAGE_INIT_FAILED, + stage = "ecstore_new", + error = ?err, + "Embedded storage initialization failed" + ); + return Err(Error::other(format!("ECStore: {err}"))); + } + }; init_embedded_startup_storage_global_config(store.clone()).await?; readiness.mark_stage(SystemStage::StorageReady); @@ -324,7 +332,9 @@ fn global_config_retry_exhausted(retry_count: usize) -> bool { #[cfg(test)] mod tests { - use super::{global_config_retry_exhausted, storage_pool_has_host_failure_risk}; + use super::{global_config_retry_exhausted, init_embedded_startup_storage_foundation, storage_pool_has_host_failure_risk}; + use crate::storage_api::startup::storage::{InstanceContext, bootstrap_instance_ctx}; + use std::sync::Arc; #[test] fn reports_host_failure_risk_only_for_multi_drive_sets() { @@ -333,6 +343,45 @@ mod tests { assert!(storage_pool_has_host_failure_risk(2)); } + // Phase 5 follow-up (backlog#1052): the embedded storage foundation writes + // its topology (endpoints, erasure kind) into the explicitly passed + // instance context, leaving the process bootstrap context untouched — the + // seam a future second embedded server needs to avoid the write-once + // panics on shared startup state. + #[tokio::test] + async fn embedded_foundation_writes_land_on_passed_instance_ctx() { + let temp_dir = tempfile::tempdir().expect("create temp volume"); + let volume = temp_dir.path().display().to_string(); + let instance_ctx = Arc::new(InstanceContext::new()); + + init_embedded_startup_storage_foundation("127.0.0.1:29123", &[volume], &instance_ctx) + .await + .expect("embedded storage foundation should initialize"); + + assert!( + instance_ctx.endpoints().is_some(), + "the passed context must hold the parsed endpoint topology" + ); + assert!( + instance_ctx.is_erasure_sd().await, + "a single local volume must record single-drive erasure on the passed context" + ); + + let bootstrap = bootstrap_instance_ctx(); + assert!( + !Arc::ptr_eq(&instance_ctx, &bootstrap), + "the test context must be distinct from the process bootstrap context" + ); + assert!( + bootstrap.endpoints().is_none(), + "the bootstrap context must not absorb topology written to an explicit context" + ); + assert!( + !bootstrap.is_erasure_sd().await, + "the bootstrap context must not absorb the erasure kind written to an explicit context" + ); + } + #[test] fn global_config_retry_limit_matches_startup_policy() { assert!(!global_config_retry_exhausted(15)); diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index e9edde377..b8b550949 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -51,28 +51,28 @@ pub(crate) use storage_api::{ DEFAULT_READ_BUFFER_SIZE, DailyAllTierStats, DeleteOptions, DiskError, DiskInfo, DiskInfoOptions, DiskResult, DiskStore, DynReader, DynReplicationPool, ECStore, Endpoint, EndpointServerPools, Error, EventArgs, ExpiryState, FS, FileInfoVersions, FileReader, FileWriter, GetObjectReader, HashReader, LocalPeerS3Client, MetricType, NotificationSys, OBJECT_LOCK_CONFIG, - ObjectInfo, ObjectLockBlockReason, ObjectOptions, ObjectPartInfo, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PolicySys, - PoolEndpoints, PutObjReader, QuotaError, RUSTFS_META_BUCKET, RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, - RenameDataResp, ReplicationStats, ReplicationStatusType, Result, SERVICE_SIGNAL_REFRESH_CONFIG, - SERVICE_SIGNAL_RELOAD_DYNAMIC, SetupType, StorageDeletedObject, StorageDiskRpcExt, StorageError, StorageGetObjectReader, - StorageObjectInfo, StorageObjectOptions, StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, - StorageReplicationConfigExt, StorageVersioningConfigExt, TONIC_RPC_PREFIX, TierConfigMgr, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, WorkloadAdmissionSnapshotProviderRef, WriteEncryption, WritePlan, access_consumer, add_object_lock_years, - all_local_disk, all_local_disk_path, check_retention_for_modification, collect_local_metrics, compression_metadata_value, - contract, decode_tags, decode_tags_to_map, delete_bucket_metadata_config, disk_drive_path, disk_endpoint, ecfs_consumer, - ecfs_extend_consumer, ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client, ecstore_cluster, ecstore_compression, - ecstore_config, ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event, ecstore_layout, ecstore_metrics, - ecstore_notification, ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk, ecstore_storage, ecstore_tier, - encode_tags, find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config, get_bucket_logging_config, - get_bucket_metadata, get_bucket_notification_config, get_bucket_object_lock_config, get_bucket_policy_raw, - get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config, - get_local_server_property, get_lock_acquire_timeout, get_public_access_block_config, head_prefix_consumer, helper_consumer, - init_background_replication, init_bucket_metadata_sys, init_ecstore_config, init_local_disks, init_lock_clients, + ObjectInfo, ObjectLockBlockReason, ObjectOptions, ObjectPartInfo, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PolicySys, PutObjReader, + QuotaError, RUSTFS_META_BUCKET, RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, + ReplicationStats, ReplicationStatusType, Result, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, + StorageDeletedObject, StorageDiskRpcExt, StorageError, StorageGetObjectReader, StorageObjectInfo, StorageObjectOptions, + StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageReplicationConfigExt, StorageVersioningConfigExt, + TONIC_RPC_PREFIX, TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef, + WriteEncryption, WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path, + check_retention_for_modification, collect_local_metrics, compression_metadata_value, contract, decode_tags, + decode_tags_to_map, delete_bucket_metadata_config, disk_drive_path, disk_endpoint, ecfs_consumer, ecfs_extend_consumer, + ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client, ecstore_cluster, ecstore_compression, ecstore_config, + ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event, ecstore_layout, ecstore_metrics, ecstore_notification, + ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk, ecstore_storage, ecstore_tier, encode_tags, + find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config, get_bucket_logging_config, get_bucket_metadata, + get_bucket_notification_config, get_bucket_object_lock_config, get_bucket_policy_raw, get_bucket_replication_config, + get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config, get_local_server_property, + get_lock_acquire_timeout, get_public_access_block_config, head_prefix_consumer, helper_consumer, init_background_replication, + init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients, is_all_buckets_not_found, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class, - load_bucket_metadata, options_consumer, prewarm_local_disk_id_map, read_config, record_replication_proxy, rpc_consumer, - runtime_sources_consumer, s3_api_consumer, save_config, serialize, set_bucket_metadata, table_catalog_path_hash, to_s3s_etag, - topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config, - try_migrate_server_config, update_bucket_metadata_config, verify_rpc_signature, wrap_reader, + load_bucket_metadata, options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, + rpc_consumer, runtime_sources_consumer, s3_api_consumer, save_config, serialize, set_bucket_metadata, + table_catalog_path_hash, to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, + try_migrate_iam_config, try_migrate_server_config, update_bucket_metadata_config, verify_rpc_signature, wrap_reader, }; #[cfg(test)] diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 30012fb0f..659ab6fa1 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -389,15 +389,14 @@ pub(crate) mod ecstore_event { pub(crate) mod ecstore_global { pub(crate) use rustfs_ecstore::api::global::{ - set_global_endpoints, set_global_region, set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, - update_erasure_type, + set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, }; } pub(crate) mod ecstore_runtime { pub(crate) use rustfs_ecstore::api::runtime::{ - boot_time, bucket_monitor, deployment_id, endpoint_pools, global_lock_client, global_lock_clients, - global_tier_config_mgr, region, rustfs_port, setup_is_dist_erasure, + InstanceContext, boot_time, bootstrap_ctx, bucket_monitor, deployment_id, endpoint_pools, global_lock_client, + global_lock_clients, global_tier_config_mgr, region, rustfs_port, setup_is_dist_erasure, }; } @@ -450,9 +449,11 @@ pub(crate) mod ecstore_set_disk { } pub(crate) mod ecstore_storage { + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::storage::init_local_disks; pub(crate) use rustfs_ecstore::api::storage::{ - ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_lock_clients, - prewarm_local_disk_id_map, + ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks_with_instance_ctx, + init_lock_clients, prewarm_local_disk_id_map_with_instance_ctx, }; } @@ -516,13 +517,15 @@ pub(crate) type FileReader = ecstore_disk::FileReader; pub(crate) type FileWriter = ecstore_disk::FileWriter; pub(crate) type FS = super::ecfs::FS; pub(crate) type HashReader = ecstore_rio::HashReader; +pub(crate) type InstanceContext = ecstore_runtime::InstanceContext; pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client; pub(crate) type MetricType = ecstore_metrics::MetricType; pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo; pub(crate) type ObjectLockBlockReason = ecstore_bucket::object_lock::objectlock_sys::ObjectLockBlockReason; pub(crate) type ObjectStoreResolver = dyn Fn() -> Option> + Send + Sync + 'static; -pub(crate) type PolicySys = ecstore_bucket::policy_sys::PolicySys; +#[cfg(test)] pub(crate) type PoolEndpoints = ecstore_layout::PoolEndpoints; +pub(crate) type PolicySys = ecstore_bucket::policy_sys::PolicySys; pub(crate) type WorkloadAdmissionSnapshotProviderRef = rustfs_ecstore::WorkloadAdmissionSnapshotProviderRef; pub(crate) type QuotaError = ecstore_bucket::quota::QuotaError; pub(crate) type RawFileInfo = rustfs_filemeta::RawFileInfo; @@ -535,7 +538,6 @@ pub(crate) type OldCurrentSize = ecstore_disk::OldCurrentSize; pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp; pub(crate) type ReplicationStatusType = ecstore_bucket::replication::ReplicationStatusType; pub(crate) type ReplicationStats = StorageReplicationStatsHandle; -pub(crate) type SetupType = ecstore_layout::SetupType; pub(crate) type StorageError = ecstore_error::StorageError; pub(crate) type TierConfigMgr = ecstore_tier::TierConfigMgr; pub(crate) use ecstore_disk::validate_batch_read_version_item_count; @@ -757,10 +759,26 @@ pub(crate) fn global_config_init_error_is_deterministic(err: &Error) -> bool { ecstore_config::com::is_server_config_corrupt_error(err) } +pub(crate) async fn init_local_disks_with_instance_ctx( + instance_ctx: &Arc, + endpoint_pools: EndpointServerPools, +) -> Result<()> { + ecstore_storage::init_local_disks_with_instance_ctx(instance_ctx, endpoint_pools).await +} + +/// Test-only legacy entry: registers disks into the ambient (process-default) +/// context, matching what pre-#1052 production startup did. +#[cfg(test)] pub(crate) async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> { ecstore_storage::init_local_disks(endpoint_pools).await } +/// The process-level bootstrap instance context that single-instance startup +/// threads through the storage foundation (Phase 5 follow-up, backlog#1052). +pub(crate) fn bootstrap_instance_ctx() -> Arc { + ecstore_runtime::bootstrap_ctx() +} + pub(crate) fn init_lock_clients(endpoint_pools: EndpointServerPools) { ecstore_storage::init_lock_clients(endpoint_pools); } @@ -773,8 +791,8 @@ pub(crate) async fn read_config(api: Arc, file: &str) -> Result ecstore_config::com::read_config(api, file).await } -pub(crate) async fn prewarm_local_disk_id_map() { - ecstore_storage::prewarm_local_disk_id_map().await; +pub(crate) async fn prewarm_local_disk_id_map_with_instance_ctx(instance_ctx: &Arc) { + ecstore_storage::prewarm_local_disk_id_map_with_instance_ctx(instance_ctx).await; } pub(crate) fn replication_queue_current_count() -> Option { @@ -793,14 +811,6 @@ pub(crate) fn shutdown_background_monitors() { rustfs_ecstore::shutdown_background_monitors(); } -pub(crate) fn set_global_endpoints(endpoints: Vec) { - ecstore_global::set_global_endpoints(endpoints); -} - -pub(crate) fn set_global_region(region: s3s::region::Region) { - ecstore_global::set_global_region(region); -} - pub(crate) fn set_global_rustfs_port(value: u16) { ecstore_global::set_global_rustfs_port(value); } @@ -815,10 +825,6 @@ pub(crate) async fn try_migrate_server_config(store: Arc) { ecstore_config::try_migrate_server_config(store, Some(decrypt_fn)).await; } -pub(crate) async fn update_erasure_type(setup_type: SetupType) { - ecstore_global::update_erasure_type(setup_type).await; -} - pub(crate) trait StorageDiskRpcExt { async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult; async fn delete_volume(&self, volume: &str, force_delete: bool) -> DiskResult<()>; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 4f83c6611..19bb24bc2 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -211,7 +211,7 @@ pub(crate) mod startup { } pub(crate) mod runtime_sources { - pub(crate) use crate::storage::storage_api::{DynReplicationPool, set_global_region, set_global_rustfs_port}; + pub(crate) use crate::storage::storage_api::{DynReplicationPool, InstanceContext, set_global_rustfs_port}; } pub(crate) mod services { @@ -226,9 +226,10 @@ pub(crate) mod startup { pub(crate) mod storage { pub(crate) use crate::storage::storage_api::{ - ECStore, EndpointServerPools, global_config_init_error_is_deterministic, init_background_replication, - init_compression_total_memory_from_backend, init_ecstore_config, init_global_config_sys, init_local_disks, - init_lock_clients, prewarm_local_disk_id_map, set_global_endpoints, try_migrate_server_config, update_erasure_type, + ECStore, EndpointServerPools, InstanceContext, bootstrap_instance_ctx, global_config_init_error_is_deterministic, + init_background_replication, init_compression_total_memory_from_backend, init_ecstore_config, init_global_config_sys, + init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map_with_instance_ctx, + try_migrate_server_config, }; } }