From c05c2cb24b6967b688699942348fc4cfd0900414 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 17 Jul 2026 15:28:26 +0800 Subject: [PATCH] fix(ecstore): resolve storage parity per pool --- Cargo.lock | 1 + crates/ecstore/Cargo.toml | 1 + crates/ecstore/src/api/mod.rs | 3 +- crates/ecstore/src/config/com.rs | 127 ++++-- crates/ecstore/src/config/mod.rs | 170 ++++++- crates/ecstore/src/config/storageclass.rs | 517 ++++++++++++++++++---- crates/ecstore/src/runtime/sources.rs | 22 +- crates/ecstore/src/store/init.rs | 179 ++++++-- crates/ecstore/src/store/init_format.rs | 18 +- rustfs/src/startup_storage.rs | 82 +++- 10 files changed, 936 insertions(+), 184 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f4884dc5a..62a20c1f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9055,6 +9055,7 @@ name = "rustfs-ecstore" version = "1.0.0-beta.10" dependencies = [ "aes-gcm", + "arc-swap", "async-channel", "async-recursion", "async-trait", diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index c08f0a248..0b330a904 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -61,6 +61,7 @@ rustfs-kms.workspace = true rustfs-s3-types = { workspace = true } rustfs-data-usage.workspace = true rustfs-object-capacity.workspace = true +arc-swap.workspace = true async-trait.workspace = true bytes = { workspace = true, features = ["serde"] } byteorder = { workspace = true } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 84e6b9c76..0694765d7 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -247,7 +247,8 @@ pub mod config { CLASS_RRS, CLASS_STANDARD, Config, DEEP_ARCHIVE, DEFAULT_INLINE_BLOCK, DEFAULT_KVS, DEFAULT_RRS_PARITY, EXPRESS_ONEZONE, GLACIER, GLACIER_IR, INLINE_BLOCK, INLINE_BLOCK_ENV, INTELLIGENT_TIERING, MIN_PARITY_DRIVES, ONEZONE_IA, OPTIMIZE, OPTIMIZE_ENV, OUTPOSTS, RRS, RRS_ENV, SCHEME_PREFIX, SNOW, STANDARD, STANDARD_ENV, STANDARD_IA, - StorageClass, default_parity_count, lookup_config, parse_storage_class, validate_parity, validate_parity_inner, + StorageClass, default_parity_count, lookup_config, lookup_config_for_pools, parse_storage_class, validate_parity, + validate_parity_inner, }; } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 5733b26f3..d6a98858d 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -425,8 +425,7 @@ async fn new_and_save_server_config(api: Arc) -> Result where S: EcstoreObjectIO + StorageAdminApi, { - let mut cfg = new_server_config(); - lookup_configs(&mut cfg, api.clone()).await; + let cfg = new_server_config(); save_server_config(api, &cfg).await?; Ok(cfg) @@ -1245,9 +1244,7 @@ where let cfg = if runtime_sources::first_cluster_node_is_local().await { new_and_save_server_config(api.clone()).await? } else { - let mut cfg = new_server_config(); - lookup_configs(&mut cfg, api).await; - cfg + new_server_config() }; warn!("Configuration initialization complete ({})", context); Ok(cfg) @@ -1545,14 +1542,11 @@ where save_config(api, &config_file, data).await } -pub async fn lookup_configs(cfg: &mut Config, api: Arc) +pub async fn lookup_configs(cfg: &mut Config, api: Arc) -> Result<()> where S: StorageAdminApi, { - // TODO: from etcd - if let Err(err) = apply_dynamic_config(cfg, api).await { - error!("apply_dynamic_config err {:?}", &err); - } + apply_dynamic_config(cfg, api).await } async fn apply_dynamic_config(cfg: &mut Config, api: Arc) -> Result<()> @@ -1569,24 +1563,34 @@ where async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: Arc, subsys: &str) -> Result<()> where S: StorageAdminApi, +{ + apply_dynamic_config_for_sub_sys_with( + cfg, + api, + subsys, + storageclass::lookup_config_for_pools, + runtime_sources::set_storage_class_config, + ) + .await +} + +async fn apply_dynamic_config_for_sub_sys_with( + cfg: &mut Config, + api: Arc, + subsys: &str, + resolve_storage_class: R, + publish_storage_class: F, +) -> Result<()> +where + S: StorageAdminApi, + R: FnOnce(&KVS, &[usize]) -> Result, + F: FnOnce(storageclass::Config), { let set_drive_counts = StorageAdminApi::set_drive_counts(api.as_ref()); if subsys == STORAGE_CLASS_SUB_SYS { let kvs = cfg.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default(); - - for (i, count) in set_drive_counts.iter().enumerate() { - match storageclass::lookup_config(&kvs, *count) { - Ok(res) => { - if i == 0 { - runtime_sources::set_storage_class_config(res); - } - } - Err(err) => { - error!("init storage class err:{:?}", &err); - break; - } - } - } + let candidate = resolve_storage_class(&kvs, &set_drive_counts)?; + publish_storage_class(candidate); } Ok(()) @@ -1595,8 +1599,8 @@ where #[cfg(test)] mod tests { use super::{ - configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config, - read_config_with_metadata, storage_class_kvs_mut, + apply_dynamic_config_for_sub_sys_with, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, + is_standard_object_server_config, lookup_configs, read_config_with_metadata, storage_class_kvs_mut, }; use crate::config::{audit, notify, oidc}; use crate::disk::endpoint::Endpoint; @@ -2977,6 +2981,7 @@ mod tests { state: Mutex, heal_replacement: Option>, heal_calls: AtomicUsize, + drive_counts: Vec, } impl RecoveryMockStore { @@ -2985,8 +2990,14 @@ mod tests { state: Mutex::new(state), heal_replacement, heal_calls: AtomicUsize::new(0), + drive_counts: vec![2], } } + + fn with_drive_counts(mut self, drive_counts: Vec) -> Self { + self.drive_counts = drive_counts; + self + } } struct ServerConfigDecryptHookGuard { @@ -3086,7 +3097,7 @@ mod tests { } fn set_drive_counts(&self) -> Vec { - vec![2] + self.drive_counts.clone() } } @@ -3127,6 +3138,68 @@ mod tests { } } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn invalid_later_pool_does_not_partially_publish_storage_class() { + temp_env::async_with_vars( + [ + (crate::config::storageclass::STANDARD_ENV, None::<&str>), + (crate::config::storageclass::RRS_ENV, None::<&str>), + (crate::config::storageclass::OPTIMIZE_ENV, None::<&str>), + (crate::config::storageclass::INLINE_BLOCK_ENV, None::<&str>), + ], + async { + let mut cfg = Config::new(); + storage_class_kvs_mut(&mut cfg) + .insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string()); + let store = + Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None).with_drive_counts(vec![4, 2])); + + let result = lookup_configs(&mut cfg, store.clone()).await; + assert!(result.is_err(), "public lookup entry must propagate an invalid later pool"); + + let mut publish_count = 0; + let result = apply_dynamic_config_for_sub_sys_with( + &mut cfg, + store, + STORAGE_CLASS_SUB_SYS, + crate::config::storageclass::lookup_config_for_pools_without_env, + |_| { + publish_count += 1; + }, + ) + .await; + assert!(result.is_err(), "invalid later pool must fail the complete candidate"); + assert_eq!(publish_count, 0, "failed reload must not publish any candidate"); + }, + ) + .await; + } + + #[tokio::test] + async fn dynamic_storage_class_publishes_one_multi_pool_snapshot() { + let mut cfg = Config::new(); + let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None).with_drive_counts(vec![4, 2])); + + let mut published = None; + let result = apply_dynamic_config_for_sub_sys_with( + &mut cfg, + store, + STORAGE_CLASS_SUB_SYS, + crate::config::storageclass::lookup_config_for_pools_without_env, + |candidate| { + published = Some(candidate); + }, + ) + .await; + + result.expect("valid multi-pool candidate should publish"); + assert_eq!( + published.and_then(|cfg| cfg.parities_for_sc(crate::config::storageclass::STANDARD)), + Some(vec![2, 1]) + ); + } + fn encrypted_current_server_config_blob() -> Vec { let mut cfg = Config::new(); let kvs = storage_class_kvs_mut(&mut cfg); diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs index 7d54da384..05bfb0e3f 100644 --- a/crates/ecstore/src/config/mod.rs +++ b/crates/ecstore/src/config/mod.rs @@ -26,6 +26,7 @@ pub mod storageclass; use crate::error::Result; use crate::store::ECStore; +use arc_swap::ArcSwap; use com::{STORAGE_CLASS_SUB_SYS, lookup_configs, read_config_without_migrate_with_recovery}; use rustfs_config::HEAL_SUB_SYS; use rustfs_config::audit::{ @@ -39,11 +40,12 @@ use rustfs_config::notify::{ use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS; use rustfs_config::server_config::{register_default_kvs, set_global_server_config}; use std::collections::HashMap; +use std::sync::Arc; use std::sync::LazyLock; -use std::sync::{Arc, RwLock}; +use tracing::warn; -pub static GLOBAL_STORAGE_CLASS: LazyLock> = - LazyLock::new(|| RwLock::new(storageclass::Config::default())); +pub static GLOBAL_STORAGE_CLASS: LazyLock> = + LazyLock::new(|| ArcSwap::from_pointee(storageclass::Config::default())); pub static GLOBAL_CONFIG_SYS: LazyLock = LazyLock::new(ConfigSys::new); pub static RUSTFS_CONFIG_PREFIX: &str = "config"; @@ -63,7 +65,7 @@ impl ConfigSys { pub async fn init(&self, api: Arc) -> Result<()> { let mut cfg = read_config_without_migrate_with_recovery(api.clone()).await?; - lookup_configs(&mut cfg, api).await; + lookup_configs(&mut cfg, api).await?; set_global_server_config(cfg); @@ -72,12 +74,166 @@ impl ConfigSys { } pub fn get_global_storage_class() -> Option { - GLOBAL_STORAGE_CLASS.read().ok().map(|guard| (*guard).clone()) + Some(GLOBAL_STORAGE_CLASS.load().as_ref().clone()) +} + +pub(crate) fn get_global_storage_class_snapshot() -> Arc { + GLOBAL_STORAGE_CLASS.load_full() +} + +fn publish_storage_class_config(target: &ArcSwap, cfg: storageclass::Config) { + let cfg = Arc::new(cfg); + target.store(cfg.clone()); + + for (pool_index, drives_per_set) in cfg.automatic_zero_parity_pools() { + warn!( + event = "storage_class_zero_redundancy", + component = "ecstore", + subsystem = "storage_class", + state = "degraded", + pool_index, + drives_per_set, + parity = 0, + "automatic storage class has no parity" + ); + } } pub fn set_global_storage_class(cfg: storageclass::Config) { - if let Ok(mut guard) = GLOBAL_STORAGE_CLASS.write() { - *guard = cfg; + publish_storage_class_config(&GLOBAL_STORAGE_CLASS, cfg); +} + +#[cfg(test)] +mod tests { + use super::publish_storage_class_config; + use crate::config::storageclass::{self, CLASS_STANDARD, STANDARD, lookup_config_for_pools_without_env}; + use arc_swap::ArcSwap; + use rustfs_config::server_config::KVS; + use std::io::{self, Write}; + use std::sync::{Arc, Mutex}; + use tracing_subscriber::fmt::MakeWriter; + + #[derive(Clone, Default)] + struct CapturedLogs { + output: Arc>>, + published: Option>>, + } + + struct CapturedWriter(CapturedLogs); + + impl Write for CapturedWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + if let Some(published) = &self.0.published { + assert_eq!( + published.load_full().parities_for_sc(STANDARD), + Some(vec![2, 0]), + "warning must be emitted after the new snapshot is visible" + ); + } + self.0.output.lock().expect("log buffer lock poisoned").extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for CapturedLogs { + type Writer = CapturedWriter; + + fn make_writer(&'a self) -> Self::Writer { + CapturedWriter(self.clone()) + } + } + + impl CapturedLogs { + fn observing(published: Arc>) -> Self { + Self { + published: Some(published), + ..Default::default() + } + } + + fn output(&self) -> String { + String::from_utf8(self.output.lock().expect("log buffer lock poisoned").clone()).expect("logs should be UTF-8") + } + } + + #[test] + fn automatic_single_disk_publish_emits_structured_warning() { + let cfg = lookup_config_for_pools_without_env(&KVS::new(), &[4, 1]).expect("automatic config should resolve"); + let target = Arc::new(ArcSwap::from_pointee(Default::default())); + let logs = CapturedLogs::observing(target.clone()); + let subscriber = tracing_subscriber::fmt().json().with_writer(logs.clone()).finish(); + + tracing::subscriber::with_default(subscriber, || publish_storage_class_config(&target, cfg)); + + assert_eq!(target.load_full().parities_for_sc(STANDARD), Some(vec![2, 0])); + let output = logs.output(); + assert_eq!( + output.matches("storage_class_zero_redundancy").count(), + 1, + "unexpected warning count: {output}" + ); + assert!(output.contains("\"level\":\"WARN\""), "unexpected warning level: {output}"); + assert!(output.contains("\"pool_index\":1"), "missing pool index: {output}"); + assert!(output.contains("\"drives_per_set\":1"), "missing drive count: {output}"); + assert!(output.contains("\"parity\":0"), "missing parity: {output}"); + } + + #[test] + fn explicit_zero_parity_publish_does_not_emit_automatic_warning() { + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:0".to_string()); + let cfg = lookup_config_for_pools_without_env(&kvs, &[1]).expect("explicit EC:0 should resolve"); + let target = ArcSwap::from_pointee(Default::default()); + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt().json().with_writer(logs.clone()).finish(); + + tracing::subscriber::with_default(subscriber, || publish_storage_class_config(&target, cfg)); + + assert_eq!(target.load_full().parities_for_sc(STANDARD), Some(vec![0])); + assert!(!logs.output().contains("storage_class_zero_redundancy")); + } + + #[test] + fn storage_class_arc_swap_never_exposes_mixed_pool_state() { + let automatic = lookup_config_for_pools_without_env(&KVS::new(), &[4, 6]).expect("automatic config should resolve"); + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:1".to_string()); + let explicit = lookup_config_for_pools_without_env(&kvs, &[4, 6]).expect("explicit config should resolve"); + let target = Arc::new(ArcSwap::from_pointee(automatic.clone())); + let old_snapshot = target.load_full(); + + let writer_target = target.clone(); + let writer = std::thread::spawn(move || { + for index in 0..2_000 { + let next = if index % 2 == 0 { automatic.clone() } else { explicit.clone() }; + writer_target.store(Arc::new(next)); + } + }); + + let readers: Vec<_> = (0..4) + .map(|_| { + let reader_target = target.clone(); + std::thread::spawn(move || { + for _ in 0..2_000 { + let pair = reader_target + .load_full() + .parities_for_sc(STANDARD) + .expect("published config must have pool topology"); + assert!(pair == [2, 3] || pair == [1, 1], "mixed pool snapshot: {pair:?}"); + } + }) + }) + .collect(); + + writer.join().expect("writer thread should complete"); + for reader in readers { + reader.join().expect("reader thread should complete"); + } + assert_eq!(old_snapshot.parities_for_sc(STANDARD), Some(vec![2, 3])); } } diff --git a/crates/ecstore/src/config/storageclass.rs b/crates/ecstore/src/config/storageclass.rs index 1016d785f..56cea2486 100644 --- a/crates/ecstore/src/config/storageclass.rs +++ b/crates/ecstore/src/config/storageclass.rs @@ -15,7 +15,7 @@ use crate::error::{Error, Result}; use rustfs_config::server_config::{KV, KVS}; use serde::{Deserialize, Serialize}; -use std::env; +use std::env::{self, VarError}; use std::sync::LazyLock; use tracing::warn; @@ -105,6 +105,13 @@ pub struct StorageClass { parity: usize, } +#[derive(Debug, Clone)] +struct PoolParity { + drives_per_set: usize, + parity: usize, + automatic: bool, +} + // Config storage class configuration #[derive(Serialize, Deserialize, Debug, Default, Clone)] pub struct Config { @@ -113,37 +120,89 @@ pub struct Config { optimize: Option, inline_block: usize, initialized: bool, + #[serde(skip)] + standard_parities: Vec, + #[serde(skip)] + rrs_parities: Vec, } impl Config { - pub fn get_parity_for_sc(&self, sc: &str) -> Option { + fn storage_class(&self, sc: &str) -> (&StorageClass, &[PoolParity]) { match sc.trim() { - RRS => { - if self.initialized { - Some(self.rrs.parity) - } else { - None - } - } - // All these storage classes use standard parity configuration - STANDARD | DEEP_ARCHIVE | EXPRESS_ONEZONE | GLACIER | GLACIER_IR | INTELLIGENT_TIERING | ONEZONE_IA | OUTPOSTS - | SNOW | STANDARD_IA => { - if self.initialized { - Some(self.standard.parity) - } else { - None - } - } - _ => { - if self.initialized { - Some(self.standard.parity) - } else { - None - } - } + RRS => (&self.rrs, &self.rrs_parities), + _ => (&self.standard, &self.standard_parities), } } + pub(crate) fn is_initialized(&self) -> bool { + self.initialized + } + + pub fn get_parity_for_sc(&self, sc: &str) -> Option { + if !self.initialized { + return None; + } + + let (storage_class, pools) = self.storage_class(sc); + let Some(first) = pools.first() else { + return Some(storage_class.parity); + }; + + pools.iter().all(|pool| pool.parity == first.parity).then_some(first.parity) + } + + /// Returns the resolved parity for a pool topology. + /// + /// A topology-bound lookup fails closed for unknown drive counts and for + /// deserialized legacy configurations that have no pool topology. Legacy + /// callers retain scalar compatibility through [`Self::get_parity_for_sc`]. + pub(crate) fn parity_for_sc(&self, sc: &str, drives_per_set: usize) -> Option { + if !self.initialized { + return None; + } + + let (_, pools) = self.storage_class(sc); + + pools + .iter() + .find(|pool| pool.drives_per_set == drives_per_set) + .map(|pool| pool.parity) + } + + pub(crate) fn parity_for_pool(&self, sc: &str, pool_index: usize, drives_per_set: usize) -> Option { + if !self.initialized { + return None; + } + + let (_, pools) = self.storage_class(sc); + + pools + .get(pool_index) + .filter(|pool| pool.drives_per_set == drives_per_set) + .map(|pool| pool.parity) + } + + pub fn parities_for_sc(&self, sc: &str) -> Option> { + if !self.initialized { + return None; + } + + let (_, pools) = self.storage_class(sc); + if pools.is_empty() { + return None; + } + + Some(pools.iter().map(|pool| pool.parity).collect()) + } + + pub(crate) fn automatic_zero_parity_pools(&self) -> impl Iterator + '_ { + self.standard_parities + .iter() + .enumerate() + .filter(|(_, pool)| pool.automatic && pool.parity == 0) + .map(|(pool_index, pool)| (pool_index, pool.drives_per_set)) + } + pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool { if shard_size < 0 { return false; @@ -180,74 +239,175 @@ impl Config { } } -pub fn lookup_config(kvs: &KVS, set_drive_count: usize) -> Result { - let standard = { - let ssc_str = { - if let Ok(ssc_str) = env::var(STANDARD_ENV) { - ssc_str - } else { - kvs.get(CLASS_STANDARD) - } - }; +#[derive(Debug, Default)] +struct StorageClassEnvOverrides { + standard: Option, + rrs: Option, + optimize: Option, + inline_block: Option, +} - if !ssc_str.is_empty() { - parse_storage_class(&ssc_str)? - } else { - StorageClass { - parity: default_parity_count(set_drive_count), - } - } - }; +impl StorageClassEnvOverrides { + fn from_process() -> Result { + Ok(Self { + standard: read_optional_env(STANDARD_ENV)?, + rrs: read_optional_env(RRS_ENV)?, + optimize: read_optional_env(OPTIMIZE_ENV)?, + inline_block: read_optional_env(INLINE_BLOCK_ENV)?, + }) + } +} - let rrs = { - let ssc_str = { - if let Ok(ssc_str) = env::var(RRS_ENV) { - ssc_str - } else { - kvs.get(CLASS_RRS) - } - }; +fn read_optional_env(name: &str) -> Result> { + parse_optional_env(name, env::var(name)) +} - if !ssc_str.is_empty() { - parse_storage_class(&ssc_str)? - } else { - StorageClass { - parity: { if set_drive_count == 1 { 0 } else { DEFAULT_RRS_PARITY } }, - } - } - }; +fn parse_optional_env(name: &str, value: std::result::Result) -> Result> { + match value { + Ok(value) => Ok(Some(value)), + Err(VarError::NotPresent) => Ok(None), + Err(VarError::NotUnicode(_)) => Err(Error::other(format!("{name} contains non-Unicode data"))), + } +} - validate_parity_inner(standard.parity, rrs.parity, set_drive_count)?; +#[derive(Debug, Clone, Copy)] +enum ParityPolicy { + AutomaticStandard, + LegacyRrs, + Explicit(usize), +} - let optimize = { env::var(OPTIMIZE_ENV).ok() }; - - let inline_block = { - if let Ok(ev) = env::var(INLINE_BLOCK_ENV) { - if let Ok(block) = ev.parse::() { - if block.as_u64() as usize > DEFAULT_INLINE_BLOCK { - warn!( - "inline block value bigger than recommended max of 128KiB -> {}, performance may degrade for PUT please benchmark the changes", - block - ); +impl ParityPolicy { + fn resolve(self, drives_per_set: usize) -> usize { + match self { + Self::AutomaticStandard => default_parity_count(drives_per_set), + Self::LegacyRrs => { + if drives_per_set == 1 { + 0 + } else { + DEFAULT_RRS_PARITY } - block.as_u64() as usize - } else { - return Err(Error::other(format!("parse {INLINE_BLOCK_ENV} format failed"))); } - } else { - DEFAULT_INLINE_BLOCK + Self::Explicit(parity) => parity, } + } +} + +fn standard_policy(kvs: &KVS, value: Option) -> Result { + match value { + Some(value) if value.is_empty() => Ok(ParityPolicy::AutomaticStandard), + Some(value) => Ok(ParityPolicy::Explicit(parse_storage_class(&value)?.parity)), + None => { + let value = kvs.get(CLASS_STANDARD); + if value.is_empty() { + Ok(ParityPolicy::AutomaticStandard) + } else { + Ok(ParityPolicy::Explicit(parse_storage_class(&value)?.parity)) + } + } + } +} + +fn rrs_policy(kvs: &KVS, value: Option) -> Result { + match value { + Some(value) if value.is_empty() => Ok(ParityPolicy::LegacyRrs), + Some(value) => Ok(ParityPolicy::Explicit(parse_storage_class(&value)?.parity)), + None => { + let value = kvs.get(CLASS_RRS); + if value.is_empty() || value == "EC:1" { + Ok(ParityPolicy::LegacyRrs) + } else { + Ok(ParityPolicy::Explicit(parse_storage_class(&value)?.parity)) + } + } + } +} + +pub fn lookup_config_for_pools(kvs: &KVS, set_drive_counts: &[usize]) -> Result { + lookup_config_for_pools_with_env(kvs, set_drive_counts, StorageClassEnvOverrides::from_process()?) +} + +fn lookup_config_for_pools_with_env( + kvs: &KVS, + set_drive_counts: &[usize], + overrides: StorageClassEnvOverrides, +) -> Result { + if set_drive_counts.is_empty() { + return Err(Error::other("storage class requires at least one pool")); + } + + let standard_policy = standard_policy(kvs, overrides.standard)?; + let rrs_policy = rrs_policy(kvs, overrides.rrs)?; + let mut standard_parities = Vec::with_capacity(set_drive_counts.len()); + let mut rrs_parities = Vec::with_capacity(set_drive_counts.len()); + + for (pool_index, &drives_per_set) in set_drive_counts.iter().enumerate() { + let standard_parity = standard_policy.resolve(drives_per_set); + let rrs_parity = rrs_policy.resolve(drives_per_set); + validate_parity_inner(standard_parity, rrs_parity, drives_per_set).map_err(|err| { + Error::other(format!( + "storage class validation failed for pool {pool_index} ({drives_per_set} drives): {err}" + )) + })?; + standard_parities.push(PoolParity { + drives_per_set, + parity: standard_parity, + automatic: matches!(standard_policy, ParityPolicy::AutomaticStandard), + }); + rrs_parities.push(PoolParity { + drives_per_set, + parity: rrs_parity, + automatic: matches!(rrs_policy, ParityPolicy::LegacyRrs), + }); + } + + let optimize = overrides.optimize; + let inline_block = if let Some(value) = overrides.inline_block { + let block = value + .parse::() + .map_err(|_| Error::other(format!("parse {INLINE_BLOCK_ENV} format failed")))?; + let inline_block = usize::try_from(block.as_u64()) + .map_err(|_| Error::other(format!("{INLINE_BLOCK_ENV} exceeds the platform size limit")))?; + if inline_block > DEFAULT_INLINE_BLOCK { + warn!( + event = "storage_class_inline_block_large", + component = "ecstore", + subsystem = "storage_class", + state = "configured", + inline_block, + recommended_max = DEFAULT_INLINE_BLOCK, + "storage class inline block exceeds recommendation" + ); + } + inline_block + } else { + DEFAULT_INLINE_BLOCK }; Ok(Config { - standard, - rrs, + standard: StorageClass { + parity: standard_parities[0].parity, + }, + rrs: StorageClass { + parity: rrs_parities[0].parity, + }, optimize, inline_block, initialized: true, + standard_parities, + rrs_parities, }) } +pub fn lookup_config(kvs: &KVS, set_drive_count: usize) -> Result { + lookup_config_for_pools(kvs, &[set_drive_count]) +} + +#[cfg(test)] +pub(crate) fn lookup_config_for_pools_without_env(kvs: &KVS, set_drive_counts: &[usize]) -> Result { + lookup_config_for_pools_with_env(kvs, set_drive_counts, StorageClassEnvOverrides::default()) +} + pub fn parse_storage_class(env: &str) -> Result { let s: Vec<&str> = env.split(':').collect(); @@ -281,6 +441,10 @@ pub fn validate_parity(ss_parity: usize, set_drive_count: usize) -> Result<()> { // ))); // } + if set_drive_count == 0 { + return Err(Error::other("set drive count must be greater than zero")); + } + if ss_parity > set_drive_count / 2 { return Err(Error::other(format!( "parity {} should be less than or equal to {}", @@ -310,22 +474,24 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun // ))); // } - if set_drive_count > 2 { - if ss_parity > set_drive_count / 2 { - return Err(Error::other(format!( - "Standard storage class parity {} should be less than or equal to {}", - ss_parity, - set_drive_count / 2 - ))); - } + if set_drive_count == 0 { + return Err(Error::other("set drive count must be greater than zero")); + } - if rrs_parity > set_drive_count / 2 { - return Err(Error::other(format!( - "Reduced redundancy storage class parity {} should be less than or equal to {}", - rrs_parity, - set_drive_count / 2 - ))); - } + if ss_parity > set_drive_count / 2 { + return Err(Error::other(format!( + "Standard storage class parity {} should be less than or equal to {}", + ss_parity, + set_drive_count / 2 + ))); + } + + if rrs_parity > set_drive_count / 2 { + return Err(Error::other(format!( + "Reduced redundancy storage class parity {} should be less than or equal to {}", + rrs_parity, + set_drive_count / 2 + ))); } if ss_parity > 0 && rrs_parity > 0 && ss_parity < rrs_parity { @@ -340,6 +506,171 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun mod tests { use super::*; + fn no_env_overrides() -> StorageClassEnvOverrides { + StorageClassEnvOverrides::default() + } + + #[test] + fn automatic_parity_is_resolved_per_pool() { + let cfg = lookup_config_for_pools_with_env(&KVS::new(), &[4, 2], no_env_overrides()) + .expect("automatic storage class should support heterogeneous pools"); + + assert_eq!(cfg.parity_for_sc(STANDARD, 4), Some(2)); + assert_eq!(cfg.parity_for_sc(STANDARD, 2), Some(1)); + assert_eq!(cfg.parity_for_sc(RRS, 4), Some(1)); + assert_eq!(cfg.parity_for_sc(RRS, 2), Some(1)); + assert_eq!(cfg.get_parity_for_sc(STANDARD), None, "heterogeneous parity has no truthful scalar"); + + let cfg = lookup_config_for_pools_with_env(&KVS::new(), &[4, 6], no_env_overrides()) + .expect("automatic parity should be valid for both pools"); + assert_eq!(cfg.parity_for_sc(STANDARD, 4), Some(2)); + assert_eq!(cfg.parity_for_sc(STANDARD, 6), Some(3)); + } + + #[test] + fn automatic_single_disk_pool_uses_zero_parity() { + let cfg = lookup_config_for_pools_with_env(&KVS::new(), &[4, 1], no_env_overrides()) + .expect("automatic single-disk storage should remain supported"); + + assert_eq!(cfg.parity_for_sc(STANDARD, 4), Some(2)); + assert_eq!(cfg.parity_for_sc(STANDARD, 1), Some(0)); + assert_eq!(cfg.parity_for_sc(RRS, 1), Some(0)); + } + + #[test] + fn explicit_standard_parity_is_validated_against_every_pool() { + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:2".to_string()); + + let err = lookup_config_for_pools_with_env(&kvs, &[4, 2], no_env_overrides()) + .expect_err("EC:2 must be rejected by the two-drive pool"); + assert!( + err.to_string().contains("pool 1") && err.to_string().contains("2 drives"), + "error must identify the rejecting pool: {err}" + ); + + kvs.insert(CLASS_STANDARD.to_string(), "EC:1".to_string()); + let cfg = lookup_config_for_pools_with_env(&kvs, &[4, 2], no_env_overrides()).expect("EC:1 is valid for both pools"); + assert_eq!(cfg.parity_for_sc(STANDARD, 4), Some(1)); + assert_eq!(cfg.parity_for_sc(STANDARD, 2), Some(1)); + assert_eq!(cfg.get_parity_for_sc(STANDARD), Some(1)); + } + + #[test] + fn environment_rrs_value_is_explicit_but_persisted_default_is_legacy() { + let persisted_default = lookup_config_for_pools_with_env(&DEFAULT_KVS, &[1], no_env_overrides()) + .expect("persisted default RRS EC:1 should retain single-disk compatibility"); + assert_eq!(persisted_default.parity_for_sc(RRS, 1), Some(0)); + + let env = StorageClassEnvOverrides { + rrs: Some("EC:1".to_string()), + ..Default::default() + }; + let err = lookup_config_for_pools_with_env(&DEFAULT_KVS, &[1], env) + .expect_err("environment RRS EC:1 is explicit and must fail on a single disk"); + assert!(err.to_string().contains("pool 0") && err.to_string().contains("1 drives")); + } + + #[test] + fn explicit_environment_standard_parity_is_not_clamped() { + let env = StorageClassEnvOverrides { + standard: Some("EC:2".to_string()), + ..Default::default() + }; + let err = lookup_config_for_pools_with_env(&KVS::new(), &[4, 2], env) + .expect_err("explicit environment parity must not be clamped for a smaller pool"); + assert!(err.to_string().contains("pool 1") && err.to_string().contains("2 drives")); + } + + #[test] + fn empty_environment_values_retain_automatic_semantics() { + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:2".to_string()); + kvs.insert(CLASS_RRS.to_string(), "EC:2".to_string()); + let env = StorageClassEnvOverrides { + standard: Some(String::new()), + rrs: Some(String::new()), + ..Default::default() + }; + + let cfg = lookup_config_for_pools_with_env(&kvs, &[4, 2], env) + .expect("empty environment values should override persisted parity with automatic defaults"); + assert_eq!(cfg.parities_for_sc(STANDARD), Some(vec![2, 1])); + assert_eq!(cfg.parities_for_sc(RRS), Some(vec![1, 1])); + } + + #[test] + fn explicit_zero_parity_is_not_reported_as_automatic() { + let env = StorageClassEnvOverrides { + standard: Some("EC:0".to_string()), + ..Default::default() + }; + + let cfg = lookup_config_for_pools_with_env(&KVS::new(), &[1], env) + .expect("explicit zero parity should remain valid for a single-drive pool"); + assert_eq!(cfg.parities_for_sc(STANDARD), Some(vec![0])); + assert_eq!(cfg.automatic_zero_parity_pools().count(), 0); + } + + #[cfg(unix)] + #[test] + fn non_unicode_environment_value_fails_closed() { + use std::ffi::OsString; + use std::os::unix::ffi::OsStringExt; + + let err = parse_optional_env(STANDARD_ENV, Err(VarError::NotUnicode(OsString::from_vec(vec![0xff])))) + .expect_err("non-Unicode parity must fail closed"); + assert!(err.to_string().contains(STANDARD_ENV)); + } + + #[test] + fn environment_standard_override_can_rescue_persisted_config() { + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:2".to_string()); + let env = StorageClassEnvOverrides { + standard: Some("EC:1".to_string()), + ..Default::default() + }; + + let cfg = lookup_config_for_pools_with_env(&kvs, &[4, 2], env) + .expect("environment override should replace the persisted parity before validation"); + assert_eq!(cfg.parity_for_pool(STANDARD, 0, 4), Some(1)); + assert_eq!(cfg.parity_for_pool(STANDARD, 1, 2), Some(1)); + } + + #[test] + fn invalid_or_unknown_pool_topology_fails_closed() { + assert!(lookup_config_for_pools_with_env(&KVS::new(), &[], no_env_overrides()).is_err()); + + let err = lookup_config_for_pools_with_env(&KVS::new(), &[4, 0], no_env_overrides()) + .expect_err("zero-drive pools must be rejected"); + assert!(err.to_string().contains("pool 1") && err.to_string().contains("0 drives")); + + let cfg = + lookup_config_for_pools_with_env(&KVS::new(), &[4, 2], no_env_overrides()).expect("valid topology should resolve"); + assert_eq!(cfg.parity_for_pool(STANDARD, 2, 6), None); + assert_eq!(cfg.parity_for_pool(STANDARD, 1, 6), None); + assert_eq!(cfg.parity_for_pool(STANDARD, 0, 2), None); + assert_eq!(cfg.parity_for_pool(STANDARD, 1, 4), None); + assert_eq!(cfg.parity_for_sc(STANDARD, 6), None); + } + + #[test] + fn resolved_pool_state_is_not_serialized() { + let cfg = + lookup_config_for_pools_with_env(&KVS::new(), &[4, 2], no_env_overrides()).expect("valid topology should resolve"); + let encoded = serde_json::to_string(&cfg).expect("config should serialize"); + assert!(!encoded.contains("standard_parities")); + assert!(!encoded.contains("rrs_parities")); + + let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize"); + assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2)); + assert_eq!(decoded.parity_for_pool(STANDARD, 0, 4), None); + assert_eq!(decoded.parity_for_sc(STANDARD, 4), None); + assert_eq!(decoded.parities_for_sc(STANDARD), None); + assert!(validate_parity(0, 0).is_err()); + } + #[test] fn lookup_config_reads_rrs_from_class_rrs_key() { // Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of @@ -348,7 +679,7 @@ mod tests { kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string()); kvs.insert(CLASS_RRS.to_string(), "EC:2".to_string()); - let cfg = lookup_config(&kvs, 8).expect("lookup should succeed"); + let cfg = lookup_config_for_pools_without_env(&kvs, &[8]).expect("lookup should succeed"); assert_eq!(cfg.standard.parity, 4, "standard parity should be 4"); assert_eq!(cfg.rrs.parity, 2, "rrs parity should be 2"); } @@ -360,7 +691,7 @@ mod tests { kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string()); kvs.insert(RRS.to_string(), "EC:2".to_string()); - let cfg = lookup_config(&kvs, 8).expect("lookup should succeed"); + let cfg = lookup_config_for_pools_without_env(&kvs, &[8]).expect("lookup should succeed"); assert_eq!(cfg.standard.parity, 4); assert_eq!( cfg.rrs.parity, DEFAULT_RRS_PARITY, diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 64e988767..4d00904e2 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -25,7 +25,7 @@ use crate::{ bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}, bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys}, bucket::replication::{DynReplicationPool, ReplicationStats}, - config::{get_global_storage_class, set_global_storage_class, storageclass}, + config::{get_global_storage_class, get_global_storage_class_snapshot, set_global_storage_class, storageclass}, disk::{DiskAPI, DiskOption, DiskStore, new_disk}, error::Result, layout::endpoints::{EndpointServerPools, SetupType}, @@ -239,23 +239,11 @@ pub(crate) fn ensure_test_rpc_secret() { } pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option { - get_global_storage_class().and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default())) -} - -pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option, Option) { - if let Some(sc) = get_global_storage_class() { - let standard = sc - .get_parity_for_sc(storageclass::CLASS_STANDARD) - .or(Some(default_standard_parity)); - let reduced_redundancy = sc.get_parity_for_sc(storageclass::RRS); - (standard, reduced_redundancy) - } else { - (Some(default_standard_parity), None) - } + get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default()) } pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool { - get_global_storage_class().is_some_and(|sc| sc.should_inline(shard_size, versioned)) + get_global_storage_class_snapshot().should_inline(shard_size, versioned) } pub(crate) fn deployment_upload_id(upload_id: &str) -> String { @@ -330,6 +318,10 @@ pub(crate) fn storage_class_config() -> Option { get_global_storage_class() } +pub(crate) fn storage_class_config_snapshot() -> Arc { + get_global_storage_class_snapshot() +} + pub(crate) fn set_storage_class_config(config: storageclass::Config) { set_global_storage_class(config); } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 1ade03736..2ca3351f5 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -18,6 +18,7 @@ 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 rustfs_config::server_config::KVS; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -31,6 +32,23 @@ fn pool_first_endpoint_is_local(pool: &crate::layout::endpoints::PoolEndpoints) pool.endpoints.as_ref().first().is_some_and(|endpoint| endpoint.is_local) } +fn startup_pool_drive_counts(endpoint_pools: &EndpointServerPools) -> Vec { + endpoint_pools.as_ref().iter().map(|pool| pool.drives_per_set).collect() +} + +fn resolve_startup_pool_defaults(endpoint_pools: &EndpointServerPools) -> Result> { + resolve_startup_pool_defaults_with(endpoint_pools, ECStore::validate_startup_storage_class) +} + +fn resolve_startup_pool_defaults_with( + endpoint_pools: &EndpointServerPools, + validate: impl FnOnce(&EndpointServerPools) -> Result<()>, +) -> Result> { + validate(endpoint_pools)?; + let drive_counts = startup_pool_drive_counts(endpoint_pools); + drive_counts.into_iter().map(ec_drives_no_config).collect() +} + fn should_resume_local_decommission(endpoints: &EndpointServerPools, idx: usize) -> Result { let pool = endpoints.as_ref().get(idx).ok_or_else(|| { Error::other(format!( @@ -163,6 +181,12 @@ async fn resume_local_decommission_after_init(store: Arc, rx: Cancellat } impl ECStore { + /// Validate topology and process storage-class overrides before any disk is opened. + pub fn validate_startup_storage_class(endpoint_pools: &EndpointServerPools) -> Result<()> { + let drive_counts = startup_pool_drive_counts(endpoint_pools); + storageclass::lookup_config_for_pools(&KVS::new(), &drive_counts).map(|_| ()) + } + #[allow(clippy::new_ret_no_self)] #[instrument(level = "debug", skip(endpoint_pools))] pub async fn new(address: SocketAddr, endpoint_pools: EndpointServerPools, ctx: CancellationToken) -> Result> { @@ -186,6 +210,11 @@ impl ECStore { ) -> Result> { // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; + // Validate topology and environment overrides before opening any disk. + // The values stored on SetDisks remain pure per-pool topology defaults; + // the runtime storage-class snapshot is published later from config. + let default_pool_parities = resolve_startup_pool_defaults(&endpoint_pools)?; + let mut deployment_id = None; // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?; @@ -222,15 +251,12 @@ impl ECStore { // debug!("endpoint_pools: {:?}", endpoint_pools); - let mut common_parity_drives = 0; - for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() { let pool_first_is_local = pool_first_endpoint_is_local(pool_eps); - if common_parity_drives == 0 { - let parity_drives = ec_drives_no_config(pool_eps.drives_per_set)?; - storageclass::validate_parity(parity_drives, pool_eps.drives_per_set)?; - common_parity_drives = parity_drives; - } + let parity_drives = default_pool_parities + .get(i) + .copied() + .ok_or_else(|| Error::other(format!("store init failed to resolve default parity for pool {i}")))?; // validate_parity(parity_count, pool_eps.drives_per_set)?; @@ -327,8 +353,7 @@ impl ECStore { } } - let sets = - Sets::new_with_instance_ctx(disks.clone(), pool_eps, &fm, i, common_parity_drives, instance_ctx.clone()).await?; + let sets = Sets::new_with_instance_ctx(disks.clone(), pool_eps, &fm, i, parity_drives, instance_ctx.clone()).await?; pools.push(sets); disk_map.insert(i, disks); @@ -493,9 +518,9 @@ impl ECStore { mod tests { use super::{ LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local, - resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init, - should_auto_start_rebalance_after_recovered_meta, should_resume_local_decommission, - should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, + resolve_startup_pool_defaults_with, resolve_store_init_stage_result, save_validated_pool_meta_for_startup, + should_auto_start_rebalance_after_init, should_auto_start_rebalance_after_recovered_meta, + should_resume_local_decommission, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, }; use crate::{ core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus}, @@ -507,7 +532,9 @@ mod tests { storage_api_contracts::{object::ObjectIO, range::HTTPRangeSpec}, }; use http::HeaderMap; + use rustfs_config::server_config::KVS; use std::{ + future::Future, io::Cursor, sync::{ Arc, @@ -784,33 +811,92 @@ mod tests { ); } - // Build a real 4-drive store over a temp dir around a fresh instance - // context — shared by the isolation tests below. + fn endpoint_pools_with_drive_counts(counts: &[usize]) -> EndpointServerPools { + EndpointServerPools::from( + counts + .iter() + .enumerate() + .map(|(pool_index, &drives_per_set)| PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set, + endpoints: Endpoints::from(Vec::new()), + cmd_line: format!("pool-{pool_index}"), + platform: String::new(), + }) + .collect::>(), + ) + } + + #[test] + fn startup_pool_defaults_are_resolved_per_pool() { + let validate = |pools: &EndpointServerPools| { + let drive_counts: Vec<_> = pools.as_ref().iter().map(|pool| pool.drives_per_set).collect(); + crate::config::storageclass::lookup_config_for_pools_without_env(&KVS::new(), &drive_counts).map(|_| ()) + }; + let defaults = resolve_startup_pool_defaults_with(&endpoint_pools_with_drive_counts(&[4, 2]), validate) + .expect("heterogeneous topology should resolve"); + assert_eq!(defaults, vec![2, 1]); + + let defaults = resolve_startup_pool_defaults_with(&endpoint_pools_with_drive_counts(&[4, 6]), validate) + .expect("heterogeneous topology should resolve"); + assert_eq!(defaults, vec![2, 3]); + } + + #[test] + fn startup_pool_defaults_validate_explicit_environment_for_every_pool() { + let validate = |pools: &EndpointServerPools| { + let drive_counts: Vec<_> = pools.as_ref().iter().map(|pool| pool.drives_per_set).collect(); + let mut kvs = KVS::new(); + kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string()); + crate::config::storageclass::lookup_config_for_pools_without_env(&kvs, &drive_counts).map(|_| ()) + }; + let err = resolve_startup_pool_defaults_with(&endpoint_pools_with_drive_counts(&[4, 2]), validate) + .expect_err("explicit EC:2 must fail before any two-drive pool I/O"); + assert!(err.to_string().contains("pool 1") && err.to_string().contains("2 drives")); + } + + async fn without_storage_class_env(future: F) -> F::Output { + temp_env::async_with_vars( + [ + (crate::config::storageclass::STANDARD_ENV, None::<&str>), + (crate::config::storageclass::RRS_ENV, None::<&str>), + (crate::config::storageclass::OPTIMIZE_ENV, None::<&str>), + (crate::config::storageclass::INLINE_BLOCK_ENV, None::<&str>), + ], + future, + ) + .await + } + + // Build a real local store over a temp dir around a fresh instance context. async fn build_isolated_test_store( temp_dir: &std::path::Path, cmd_line: &str, + pool_drive_counts: &[usize], ) -> (Arc, Arc) { - let disk_paths: Vec<_> = (1..=4).map(|i| temp_dir.join(format!("disk{i}"))).collect(); - for path in &disk_paths { - tokio::fs::create_dir_all(path).await.expect("create disk dir"); + let mut pools = Vec::with_capacity(pool_drive_counts.len()); + for (pool_index, &drives_per_set) in pool_drive_counts.iter().enumerate() { + let mut endpoints = Vec::with_capacity(drives_per_set); + for disk_index in 0..drives_per_set { + let path = temp_dir.join(format!("pool{pool_index}/disk{disk_index}")); + tokio::fs::create_dir_all(&path).await.expect("create disk dir"); + let mut endpoint = Endpoint::try_from(path.to_str().expect("disk path should be utf-8")).expect("local endpoint"); + endpoint.set_pool_index(pool_index); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_index); + endpoints.push(endpoint); + } + pools.push(PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set, + endpoints: Endpoints::from(endpoints), + cmd_line: format!("{cmd_line}-pool-{pool_index}"), + platform: "test".to_string(), + }); } - - 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: cmd_line.to_string(), - platform: "test".to_string(), - }]); + let endpoint_pools = EndpointServerPools(pools); let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) @@ -835,9 +921,11 @@ mod tests { // context, not on the process bootstrap one. This is the storage-layer // seam a future second embedded server needs to stay isolated. #[tokio::test] + #[serial_test::serial(storage_class_env)] 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; + let (instance_ctx, store) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "instance-ctx-store-graph-test", &[4])).await; assert!( Arc::ptr_eq(&store.ctx, &instance_ctx), @@ -873,16 +961,33 @@ mod tests { } } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn new_with_instance_ctx_applies_default_parity_to_each_real_pool() { + let temp_dir = tempfile::tempdir().expect("create multi-pool store dir"); + let (_, store) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "pool-parity-regression", &[4, 2])).await; + + assert_eq!(store.pools.len(), 2); + assert_eq!(store.pools[0].default_parity_count, 2); + assert_eq!(store.pools[0].disk_set[0].default_parity_count, 2); + assert_eq!(store.pools[1].default_parity_count, 1); + assert_eq!(store.pools[1].disk_set[0].default_parity_count, 1); + } + // 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] + #[serial_test::serial(storage_class_env)] 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; + let (ctx_a, store_a) = + without_storage_class_env(build_isolated_test_store(temp_a.path(), "bucket-metadata-isolation-a", &[4])).await; + let (ctx_b, store_b) = + without_storage_class_env(build_isolated_test_store(temp_b.path(), "bucket-metadata-isolation-b", &[4])).await; crate::bucket::metadata_sys::init_bucket_metadata_sys(store_a.clone(), Vec::new()).await; // The old process-global cell would panic right here. diff --git a/crates/ecstore/src/store/init_format.rs b/crates/ecstore/src/store/init_format.rs index 12d3179a5..01ecb84b6 100644 --- a/crates/ecstore/src/store/init_format.rs +++ b/crates/ecstore/src/store/init_format.rs @@ -26,7 +26,6 @@ use crate::{ layout::endpoints::Endpoints, }; use futures::future::join_all; -use rustfs_config::server_config::KVS; use std::collections::{HashMap, hash_map::Entry}; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -476,8 +475,21 @@ pub async fn save_format_file(disk: &Option, format: &Option Result { - let sc = storageclass::lookup_config(&KVS::new(), set_drive_count)?; - Ok(sc.get_parity_for_sc(storageclass::STANDARD).unwrap_or_default()) + let parity = storageclass::default_parity_count(set_drive_count); + storageclass::validate_parity(parity, set_drive_count)?; + Ok(parity) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ec_drives_no_config_uses_topology_defaults() { + assert_eq!(ec_drives_no_config(1).expect("single-drive topology should resolve"), 0); + assert_eq!(ec_drives_no_config(2).expect("two-drive topology should resolve"), 1); + assert_eq!(ec_drives_no_config(6).expect("six-drive topology should resolve"), 3); + } } // #[derive(Debug, PartialEq, thiserror::Error)] diff --git a/rustfs/src/startup_storage.rs b/rustfs/src/startup_storage.rs index 56e090ce4..2053f01cb 100644 --- a/rustfs/src/startup_storage.rs +++ b/rustfs/src/startup_storage.rs @@ -73,6 +73,20 @@ pub(crate) async fn init_startup_storage_foundation( ); }) .map_err(Error::other)?; + ECStore::validate_startup_storage_class(&endpoint_pools) + .inspect_err(|err| { + error!( + target: "rustfs::main::run", + event = EVENT_STARTUP_STORAGE_STAGE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, + stage = "storage_class_validation", + state = "failed", + error = %err, + "Storage-class validation failed" + ); + }) + .map_err(Error::other)?; enforce_unsupported_fs_policy(&endpoint_pools)?; instance_ctx.set_endpoints(endpoint_pools.clone()); @@ -118,6 +132,7 @@ pub(crate) async fn init_embedded_startup_storage_foundation( let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address, volumes.to_vec()) .await .map_err(|err| Error::other(format!("endpoints: {err}")))?; + ECStore::validate_startup_storage_class(&endpoint_pools).map_err(|err| Error::other(format!("storage class: {err}")))?; enforce_unsupported_fs_policy(&endpoint_pools).map_err(|err| Error::other(format!("unsupported fs guard: {err}")))?; instance_ctx.set_endpoints(endpoint_pools.clone()); @@ -332,8 +347,13 @@ fn global_config_retry_exhausted(retry_count: usize) -> bool { #[cfg(test)] mod tests { - use super::{global_config_retry_exhausted, init_embedded_startup_storage_foundation, storage_pool_has_host_failure_risk}; + use super::{ + global_config_retry_exhausted, init_embedded_startup_storage_foundation, init_startup_storage_foundation, + storage_pool_has_host_failure_risk, + }; + use crate::storage::storage_api::ecstore_config::storageclass::{INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV}; use crate::storage_api::startup::storage::{InstanceContext, bootstrap_instance_ctx}; + use rustfs_config::ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT; use std::sync::Arc; #[test] @@ -349,6 +369,7 @@ mod tests { // seam a future second embedded server needs to avoid the write-once // panics on shared startup state. #[tokio::test] + #[serial_test::serial(storage_class_env)] 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(); @@ -382,6 +403,65 @@ mod tests { ); } + fn heterogeneous_volume_args(temp_dir: &tempfile::TempDir) -> Vec { + let root = temp_dir.path().display(); + vec![format!("{root}/pool-a/disk{{1...4}}"), format!("{root}/pool-b/disk{{1...2}}")] + } + + async fn assert_invalid_storage_class_fails_before_foundation_mutation(embedded: bool) { + let temp_dir = tempfile::tempdir().expect("create topology root"); + let volumes = heterogeneous_volume_args(&temp_dir); + let instance_ctx = Arc::new(InstanceContext::new()); + + let result = if embedded { + init_embedded_startup_storage_foundation("127.0.0.1:29124", &volumes, &instance_ctx).await + } else { + init_startup_storage_foundation("127.0.0.1:29125", &volumes, &instance_ctx).await + }; + + let err = result.expect_err("EC:2 must be rejected by the two-drive pool before disk initialization"); + assert!( + err.to_string().contains("pool 1") && err.to_string().contains("2 drives"), + "unexpected error: {err}" + ); + assert!( + instance_ctx.endpoints().is_none(), + "invalid storage-class configuration must fail before mutating the instance topology" + ); + } + + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn startup_foundation_validates_all_pools_before_mutation() { + temp_env::async_with_vars( + [ + (STANDARD_ENV, Some("EC:2")), + (RRS_ENV, None), + (OPTIMIZE_ENV, None), + (INLINE_BLOCK_ENV, None), + (ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, None), + ], + assert_invalid_storage_class_fails_before_foundation_mutation(false), + ) + .await; + } + + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn embedded_foundation_validates_all_pools_before_mutation() { + temp_env::async_with_vars( + [ + (STANDARD_ENV, Some("EC:2")), + (RRS_ENV, None), + (OPTIMIZE_ENV, None), + (INLINE_BLOCK_ENV, None), + (ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, None), + ], + assert_invalid_storage_class_fails_before_foundation_mutation(true), + ) + .await; + } + #[test] fn global_config_retry_limit_matches_startup_policy() { assert!(!global_config_retry_exhausted(15));