From 7f5873dac8647deb5ef1fed8ce4011a10a42ed85 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 19 Jul 2026 11:06:46 +0800 Subject: [PATCH] fix(ecstore): resolve erasure parity per pool (#4801) (#5015) * fix(ecstore): add fallible erasure construction (cherry picked from commit bd148b20f7aecac2e26855740fc73945b62147f3) * fix(ecstore): resolve storage parity per pool (cherry picked from commit c05c2cb24b6967b688699942348fc4cfd0900414) * fix(ecstore): keep carved per-pool parity core self-contained on main Fixups so the cherry-picked fallible-erasure + per-pool-parity core builds standalone on current main without the excluded scope-creep commits: - runtime/sources.rs: re-add backend_storage_class_parities (removed by the per-pool commit; its rebalance caller was updated in an unrelated reporting commit that was left out). Reimplemented over the snapshot API, behavior-identical. - config/mod.rs: rename the storage-class publish test module (main independently added a mod tests, so the cherry-pick collided). - rustfs storage_api.rs + startup_storage.rs: route the storage-class ENV consts through the startup storage facade and use a local const for the erasure-set-drive-count env name, satisfying the layer/facade guardrail (main's guardrail is stricter than when the core was authored). * fix(ecstore): use struct-init in erasure test helper to satisfy clippy field_reassign_with_default The cherry-picked fallible-erasure commit's `erasure_with_invalid_dimensions` test helper built `Erasure` via `default()` then reassigned fields, which trips `clippy::field_reassign_with_default` under `-D warnings` (only surfaced by `--all-targets`, which lints test code). #4977 fixed this in a later commit that was not part of the carved core. Use struct-init with `..Default::default()`, matching #4977's final form. * fix(rustfs): gate the test-only storage-class ENV facade re-export behind cfg(test) The ENV constants (INLINE_BLOCK_ENV/OPTIMIZE_ENV/RRS_ENV/STANDARD_ENV) re-exported through the startup storage facade are only consumed by a #[cfg(test)] test in startup_storage.rs, so in a non-test lib build the re-export is unused and trips -D unused-imports under clippy --all-targets. Gate it with #[cfg(test)], matching #4977's final form. --------- Co-authored-by: cxymds --- Cargo.lock | 1 + crates/ecstore/Cargo.toml | 1 + crates/ecstore/src/api/mod.rs | 7 +- crates/ecstore/src/config/com.rs | 128 ++++- crates/ecstore/src/config/mod.rs | 170 +++++- crates/ecstore/src/config/storageclass.rs | 517 ++++++++++++++---- crates/ecstore/src/disk/error.rs | 28 +- crates/ecstore/src/disk/local.rs | 31 +- crates/ecstore/src/erasure/coding/encode.rs | 10 +- crates/ecstore/src/erasure/coding/erasure.rs | 325 ++++++++++- crates/ecstore/src/erasure/coding/mod.rs | 2 +- crates/ecstore/src/error/mod.rs | 28 +- crates/ecstore/src/runtime/sources.rs | 37 +- crates/ecstore/src/set_disk/ops/heal.rs | 3 +- crates/ecstore/src/set_disk/ops/multipart.rs | 3 +- crates/ecstore/src/set_disk/ops/object.rs | 38 +- crates/ecstore/src/set_disk/read.rs | 44 +- crates/ecstore/src/store/init.rs | 179 ++++-- crates/ecstore/src/store/init_format.rs | 18 +- rustfs/src/startup_storage.rs | 88 ++- rustfs/src/storage_api.rs | 4 + scripts/check_architecture_migration_rules.sh | 49 ++ 22 files changed, 1463 insertions(+), 248 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 12e32c157..447701e42 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9062,6 +9062,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 8a5f6446e..d51fbaa8e 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, }; } @@ -310,8 +311,8 @@ pub mod error { pub mod erasure { pub use crate::erasure::coding::{ - BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ReedSolomonEncoder, calc_shard_size, - calc_shard_size_legacy, + BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError, ReedSolomonEncoder, + calc_shard_size, calc_shard_size_legacy, }; } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index a32662f6a..f8c89871c 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -456,8 +456,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) @@ -1276,9 +1275,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) @@ -1576,14 +1573,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<()> @@ -1600,24 +1594,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(()) @@ -1626,8 +1630,9 @@ 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, read_config_preserve_empty, 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, read_config_preserve_empty, read_config_with_metadata, + storage_class_kvs_mut, }; use crate::config::{audit, notify, oidc}; use crate::disk::endpoint::Endpoint; @@ -3008,6 +3013,7 @@ mod tests { state: Mutex, heal_replacement: Option>, heal_calls: AtomicUsize, + drive_counts: Vec, } impl RecoveryMockStore { @@ -3016,8 +3022,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 { @@ -3132,7 +3144,7 @@ mod tests { } fn set_drive_counts(&self) -> Vec { - vec![2] + self.drive_counts.clone() } } @@ -3173,6 +3185,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..f1fe35811 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 storage_class_publish_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/disk/error.rs b/crates/ecstore/src/disk/error.rs index 606ec7456..bb6c07b5f 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -141,7 +141,7 @@ pub enum DiskError { ErasureReadQuorum, #[error("io error {0}")] - Io(io::Error), + Io(#[source] io::Error), #[error("source stalled")] SourceStalled, @@ -153,6 +153,12 @@ pub enum DiskError { InvalidPath, } +impl From for DiskError { + fn from(error: crate::erasure::coding::ErasureConstructionError) -> Self { + Self::Io(error.into_io_error()) + } +} + impl DiskError { pub fn other(error: E) -> Self where @@ -601,6 +607,26 @@ mod tests { use super::*; use std::collections::HashMap; + #[test] + fn other_preserves_erasure_construction_source_chain() { + use crate::erasure::coding::ErasureConstructionError; + use std::error::Error as _; + + let error = DiskError::from(ErasureConstructionError::ModernEncoder { + source: reed_solomon_erasure::Error::TooManyShards, + }); + let io_source = error.source().expect("DiskError::Io must expose its io::Error source"); + assert!(io_source.is::()); + let construction_source = io_source + .source() + .expect("io::Error must expose the erasure construction error"); + assert!(construction_source.is::()); + let encoder_source = construction_source + .source() + .expect("construction error must expose the encoder error"); + assert!(encoder_source.is::()); + } + #[test] fn test_disk_error_variants() { let errors = vec![ diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 9d0c7fca0..15ff12815 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -5962,12 +5962,13 @@ impl DiskAPI for LocalDisk { }; let erasure = &fi.erasure; - let codec_erasure = coding::Erasure::new_with_options( + let codec_erasure = coding::Erasure::try_new_with_options( erasure.data_blocks, erasure.parity_blocks, erasure.block_size, fi.uses_legacy_checksum, - ); + ) + .map_err(DiskError::from)?; for (i, part) in fi.parts.iter().enumerate() { let checksum_info = erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { @@ -13600,6 +13601,32 @@ mod test { assert!(!is_bitrot_verification_error(&io::Error::other("unrelated io failure"))); } + #[tokio::test] + async fn local_disk_verify_file_preserves_erasure_construction_error() { + use crate::erasure::coding::ErasureConstructionError; + use tempfile::tempdir; + + let root_dir = tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + let volume = "verify-volume"; + ensure_test_volume(&disk, volume).await; + + let mut file_info = FileInfo::new("invalid.bin", 2, 2); + file_info.erasure.block_size = 0; + let error = match disk.verify_file(volume, "invalid.bin", &file_info).await { + Ok(_) => panic!("invalid local-disk erasure metadata must be rejected"), + Err(error) => error, + }; + + assert!(error.to_string().contains("block_size must be greater than zero")); + let io_source = std::error::Error::source(&error).expect("DiskError::Io must expose its io::Error source"); + let construction_source = io_source + .source() + .expect("io::Error must expose the erasure construction error"); + assert!(construction_source.is::()); + } + #[tokio::test] async fn local_disk_read_file_verifier_reports_bitrot_mismatch() { use crate::erasure::coding::BitrotWriter; diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index 8509275b8..117f55da8 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -841,6 +841,12 @@ mod tests { use std::time::Duration; use tokio::io::{AsyncWrite, AsyncWriteExt}; + fn erasure_with_zero_block_size() -> Erasure { + let mut erasure = Erasure::default(); + erasure.data_shards = 1; + erasure + } + #[derive(Clone, Default)] struct DeferredCommitWriter { buffered: Vec, @@ -1500,7 +1506,7 @@ mod tests { HashAlgorithm::HighwayHash256S, ))]; - let erasure = Arc::new(Erasure::new(1, 0, 0)); + let erasure = Arc::new(erasure_with_zero_block_size()); let reader = tokio::io::BufReader::new(Cursor::new(b"payload".to_vec())); let err = erasure .encode(reader, &mut writers, 1) @@ -1660,7 +1666,7 @@ mod tests { let writer = DeferredCommitWriter::new(committed); let mut writers = vec![Some(bitrot_writer(writer, 16))]; - let erasure = Arc::new(Erasure::new(1, 0, 0)); + let erasure = Arc::new(erasure_with_zero_block_size()); let reader = tokio::io::BufReader::new(Cursor::new(b"payload".to_vec())); let err = erasure .encode_batched(reader, &mut writers, 1) diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index a3f334f9a..98a9d2003 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -25,6 +25,62 @@ use tokio::io::AsyncRead; use tracing::warn; use uuid::Uuid; +const MODERN_MAX_TOTAL_SHARDS: usize = ::ORDER; + +/// Errors returned when constructing an [`Erasure`] codec. +#[derive(Debug, thiserror::Error)] +pub enum ErasureConstructionError { + /// Erasure coding requires at least one data shard. + #[error("data_shards must be greater than zero")] + ZeroDataShards, + + /// Erasure coding requires a non-zero logical block size. + #[error("block_size must be greater than zero")] + ZeroBlockSize, + + /// The total shard count cannot be represented by `usize`. + #[error("data_shards ({data_shards}) + parity_shards ({parity_shards}) overflows usize")] + ShardCountOverflow { data_shards: usize, parity_shards: usize }, + + /// The modern GF(2^8) codec cannot represent this shard configuration. + #[error("modern codec does not support {data_shards} data shards and {parity_shards} parity shards")] + UnsupportedModernShardCount { data_shards: usize, parity_shards: usize }, + + /// The legacy SIMD codec cannot represent this shard configuration. + #[error("legacy codec does not support {data_shards} data shards and {parity_shards} parity shards")] + UnsupportedLegacyShardCount { data_shards: usize, parity_shards: usize }, + + /// The modern encoder rejected dimensions that passed the preflight checks. + #[error("failed to construct modern Reed-Solomon encoder")] + ModernEncoder { + #[source] + source: reed_solomon_erasure::Error, + }, + + /// The legacy encoder wrapper failed to initialize. + #[error("failed to construct legacy Reed-Solomon encoder")] + LegacyEncoder { + #[source] + source: io::Error, + }, +} + +#[derive(Debug, thiserror::Error)] +#[error("{source}")] +struct ErasureConstructionIoSource { + #[source] + source: ErasureConstructionError, +} + +impl ErasureConstructionError { + pub(crate) fn into_io_error(self) -> io::Error { + // `io::Error::source` forwards to the wrapped error's source rather + // than exposing the wrapped error itself. This adapter keeps the + // construction error as an observable link in the source chain. + io::Error::other(ErasureConstructionIoSource { source: self }) + } +} + /// Legacy calc_shard_size formula: (block_size.div_ceil(data_shards) + 1) & !1 /// Matches main branch and filemeta::ErasureInfo for old-version files. pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize { @@ -233,12 +289,9 @@ impl Clone for ReedSolomonEncoder { } impl ReedSolomonEncoder { - /// Create a new Reed-Solomon encoder with specified data and parity shards. - pub fn new(data_shards: usize, parity_shards: usize) -> io::Result { + fn try_new_typed(data_shards: usize, parity_shards: usize) -> Result { let encoder = if parity_shards > 0 { - ReedSolomon::new(data_shards, parity_shards) - .map_err(|e| io::Error::other(format!("Failed to create Reed-Solomon encoder: {e:?}"))) - .map(Some)? + Some(ReedSolomon::new(data_shards, parity_shards)?) } else { None }; @@ -250,6 +303,12 @@ impl ReedSolomonEncoder { }) } + /// Create a new Reed-Solomon encoder with specified data and parity shards. + pub fn new(data_shards: usize, parity_shards: usize) -> io::Result { + Self::try_new_typed(data_shards, parity_shards) + .map_err(|error| io::Error::other(format!("Failed to create Reed-Solomon encoder: {error:?}"))) + } + /// Encode data shards with parity. pub fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> { let mut shards_vec: Vec<&mut [u8]> = shards.into_vec(); @@ -464,28 +523,105 @@ impl Erasure { /// * `data_shards` - Number of data shards. /// * `parity_shards` - Number of parity shards. /// * `block_size` - Block size for each shard. + /// + /// # Panics + /// Panics when the dimensions are invalid or unsupported. RustFS production + /// paths use [`Self::try_new`] instead. pub fn new(data_shards: usize, parity_shards: usize, block_size: usize) -> Self { - Self::new_with_options(data_shards, parity_shards, block_size, false) + match Self::try_new(data_shards, parity_shards, block_size) { + Ok(erasure) => erasure, + Err(error) => panic!("invalid erasure codec configuration: {error}"), + } } /// Create a new Erasure instance with legacy format support. /// /// When `uses_legacy` is true, uses main-branch shard_size formula and reed-solomon-simd /// for decode/reconstruct (for reading and healing old-version files). + /// + /// # Panics + /// Panics when the dimensions are invalid or unsupported. RustFS production + /// paths use [`Self::try_new_with_options`] instead. pub fn new_with_options(data_shards: usize, parity_shards: usize, block_size: usize, uses_legacy: bool) -> Self { + match Self::try_new_with_options(data_shards, parity_shards, block_size, uses_legacy) { + Ok(erasure) => erasure, + Err(error) => panic!("invalid erasure codec configuration: {error}"), + } + } + + /// Tries to create a modern Erasure codec. + /// + /// # Errors + /// Returns [`ErasureConstructionError`] when dimensions are zero, + /// overflow the shard count, exceed the modern codec's supported range, or + /// the underlying encoder rejects the configuration. + pub fn try_new(data_shards: usize, parity_shards: usize, block_size: usize) -> Result { + Self::try_new_with_options(data_shards, parity_shards, block_size, false) + } + + /// Tries to create an Erasure codec using the selected format backend. + /// + /// # Errors + /// Returns [`ErasureConstructionError`] when dimensions are zero, + /// overflow the shard count, are unsupported by the selected codec, or the + /// selected encoder fails to initialize. + pub fn try_new_with_options( + data_shards: usize, + parity_shards: usize, + block_size: usize, + uses_legacy: bool, + ) -> Result { + if data_shards == 0 { + return Err(ErasureConstructionError::ZeroDataShards); + } + if block_size == 0 { + return Err(ErasureConstructionError::ZeroBlockSize); + } + + let total_shards = data_shards + .checked_add(parity_shards) + .ok_or(ErasureConstructionError::ShardCountOverflow { + data_shards, + parity_shards, + })?; + + if parity_shards > 0 { + if uses_legacy + && (total_shards > reed_solomon_simd::engine::GF_ORDER + || !reed_solomon_simd::ReedSolomonEncoder::supports(data_shards, parity_shards)) + { + return Err(ErasureConstructionError::UnsupportedLegacyShardCount { + data_shards, + parity_shards, + }); + } + if !uses_legacy && total_shards > MODERN_MAX_TOTAL_SHARDS { + return Err(ErasureConstructionError::UnsupportedModernShardCount { + data_shards, + parity_shards, + }); + } + } + let encoder = if !uses_legacy && parity_shards > 0 { - Some(ReedSolomonEncoder::new(data_shards, parity_shards).expect("operation should succeed")) + Some( + ReedSolomonEncoder::try_new_typed(data_shards, parity_shards) + .map_err(|source| ErasureConstructionError::ModernEncoder { source })?, + ) } else { None }; let legacy_encoder = if uses_legacy && parity_shards > 0 { - Some(LegacyReedSolomonEncoder::new(data_shards, parity_shards).expect("operation should succeed")) + Some( + LegacyReedSolomonEncoder::new(data_shards, parity_shards) + .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?, + ) } else { None }; - Erasure { + Ok(Erasure { data_shards, parity_shards, block_size, @@ -493,7 +629,7 @@ impl Erasure { legacy_encoder, uses_legacy, _id: Uuid::nil(), // Unused in hot paths; avoid CSPRNG syscall - } + }) } /// Encode data into data and parity shards. @@ -946,6 +1082,15 @@ mod tests { use std::task::{Context, Poll}; use tokio::io::ReadBuf; + fn erasure_with_invalid_dimensions(data_shards: usize, parity_shards: usize, block_size: usize) -> Erasure { + Erasure { + data_shards, + parity_shards, + block_size, + ..Default::default() + } + } + fn optional_shards(shards: &[Bytes]) -> Vec>> { shards.iter().map(|shard| Some(shard.to_vec())).collect() } @@ -1010,6 +1155,156 @@ mod tests { } } + #[test] + fn try_new_rejects_zero_and_overflowing_dimensions_with_typed_errors() { + let Err(zero_data) = Erasure::try_new(0, 2, 64) else { + panic!("zero data shards must be rejected"); + }; + assert!(matches!(zero_data, ErasureConstructionError::ZeroDataShards)); + + let Err(zero_block) = Erasure::try_new(2, 2, 0) else { + panic!("zero block size must be rejected"); + }; + assert!(matches!(zero_block, ErasureConstructionError::ZeroBlockSize)); + + let Err(overflow) = Erasure::try_new(usize::MAX, 1, 64) else { + panic!("overflowing shard counts must be rejected"); + }; + assert!(matches!( + overflow, + ErasureConstructionError::ShardCountOverflow { + data_shards: usize::MAX, + parity_shards: 1, + } + )); + } + + #[test] + fn try_new_rejects_backend_specific_unsupported_dimensions() { + let Err(modern) = Erasure::try_new(256, 1, 64) else { + panic!("modern dimensions beyond GF(2^8) must be rejected"); + }; + assert!(matches!( + modern, + ErasureConstructionError::UnsupportedModernShardCount { + data_shards: 256, + parity_shards: 1, + } + )); + + let Err(legacy) = Erasure::try_new_with_options(60_000, 5_000, 64, true) else { + panic!("unsupported legacy SIMD dimensions must be rejected"); + }; + assert!(matches!( + legacy, + ErasureConstructionError::UnsupportedLegacyShardCount { + data_shards: 60_000, + parity_shards: 5_000, + } + )); + } + + #[test] + fn try_new_accepts_exact_backend_shard_limits() { + let modern_data_shards = MODERN_MAX_TOTAL_SHARDS - 1; + let modern = + Erasure::try_new(modern_data_shards, 1, 64).expect("the modern codec's exact field-size boundary must remain valid"); + assert_eq!(modern.total_shard_count(), MODERN_MAX_TOTAL_SHARDS); + assert!(modern.encoder.is_some()); + assert!(modern.legacy_encoder.is_none()); + + let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER / 2; + let legacy_parity_shards = reed_solomon_simd::engine::GF_ORDER - legacy_data_shards; + let legacy = Erasure::try_new_with_options(legacy_data_shards, legacy_parity_shards, 64, true) + .expect("the legacy codec's exact field-size boundary must remain valid"); + assert_eq!(legacy.total_shard_count(), reed_solomon_simd::engine::GF_ORDER); + assert!(legacy.encoder.is_none()); + assert!(legacy.legacy_encoder.is_some()); + } + + #[test] + fn try_new_accepts_zero_parity_without_backend_limits() { + let no_parity = Erasure::try_new(2, 0, 64).expect("zero parity is a valid generic codec configuration"); + assert_eq!(no_parity.total_shard_count(), 2); + assert!(no_parity.encoder.is_none()); + assert!(no_parity.legacy_encoder.is_none()); + + let large_modern = Erasure::try_new(257, 0, 64).expect("zero parity must not inherit the modern backend shard limit"); + assert_eq!(large_modern.total_shard_count(), 257); + assert!(large_modern.encoder.is_none()); + assert!(large_modern.legacy_encoder.is_none()); + + let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER + 1; + let large_legacy = Erasure::try_new_with_options(legacy_data_shards, 0, 64, true) + .expect("zero parity must not inherit the legacy backend shard limit"); + assert_eq!(large_legacy.total_shard_count(), legacy_data_shards); + assert!(large_legacy.encoder.is_none()); + assert!(large_legacy.legacy_encoder.is_none()); + } + + #[test] + fn try_new_accepts_generic_layouts_above_storage_drive_limits() { + let erasure = Erasure::try_new(2, 3, 64).expect("generic 2+3 erasure coding must remain supported"); + assert_eq!(erasure.total_shard_count(), 5); + assert!(erasure.encoder.is_some()); + assert!(erasure.legacy_encoder.is_none()); + + let modern = Erasure::try_new(9, 8, 64).expect("the codec must not impose the storage layer's drive limit"); + assert_eq!(modern.total_shard_count(), 17); + assert!(modern.encoder.is_some()); + assert!(modern.legacy_encoder.is_none()); + + let legacy = Erasure::try_new_with_options(9, 8, 64, true) + .expect("the legacy codec must not impose the storage layer's drive limit"); + assert_eq!(legacy.total_shard_count(), 17); + assert!(legacy.encoder.is_none()); + assert!(legacy.legacy_encoder.is_some()); + } + + #[test] + fn try_new_with_options_initializes_only_the_selected_backend() { + let modern = Erasure::try_new_with_options(4, 2, 64, false).expect("modern codec should construct"); + assert!(modern.encoder.is_some()); + assert!(modern.legacy_encoder.is_none()); + + let legacy = Erasure::try_new_with_options(4, 2, 64, true).expect("legacy codec should construct"); + assert!(legacy.encoder.is_none()); + assert!(legacy.legacy_encoder.is_some()); + } + + #[test] + fn construction_errors_preserve_encoder_sources() { + let modern = ErasureConstructionError::ModernEncoder { + source: reed_solomon_erasure::Error::TooManyShards, + }; + assert!( + std::error::Error::source(&modern) + .and_then(|source| source.downcast_ref::()) + .is_some() + ); + + let legacy = ErasureConstructionError::LegacyEncoder { + source: io::Error::other("legacy encoder failure"), + }; + assert!( + std::error::Error::source(&legacy) + .and_then(|source| source.downcast_ref::()) + .is_some() + ); + } + + #[test] + #[should_panic(expected = "invalid erasure codec configuration")] + fn new_panics_on_invalid_dimensions() { + let _ = Erasure::new(0, 1, 64); + } + + #[test] + #[should_panic(expected = "invalid erasure codec configuration")] + fn new_with_options_panics_on_invalid_dimensions() { + let _ = Erasure::new_with_options(1, 1, 0, true); + } + #[test] fn has_valid_dimensions_rejects_zero_block_size_or_data_shards() { // Well-formed erasure metadata is accepted. @@ -1018,11 +1313,9 @@ mod tests { // Corrupted on-disk metadata with a zero block_size or zero data_shards // must be rejected before it reaches the shard/offset divisions. - // (parity_shards is kept 0 for the zero-data_shards cases so the - // Reed-Solomon encoder is not constructed with an invalid shard count.) - assert!(!Erasure::new(4, 2, 0).has_valid_dimensions()); - assert!(!Erasure::new(0, 0, 64).has_valid_dimensions()); - assert!(!Erasure::new(0, 0, 0).has_valid_dimensions()); + assert!(!erasure_with_invalid_dimensions(4, 2, 0).has_valid_dimensions()); + assert!(!erasure_with_invalid_dimensions(0, 0, 64).has_valid_dimensions()); + assert!(!erasure_with_invalid_dimensions(0, 0, 0).has_valid_dimensions()); } #[tokio::test] @@ -1762,7 +2055,7 @@ mod tests { use std::io::Cursor; use std::sync::{Arc, Mutex}; - let erasure = Arc::new(Erasure::new(1, 0, 0)); + let erasure = Arc::new(erasure_with_invalid_dimensions(1, 0, 0)); let mut reader = Cursor::new(b"payload".to_vec()); let observed = Arc::new(Mutex::new(None)); let observed_clone = observed.clone(); diff --git a/crates/ecstore/src/erasure/coding/mod.rs b/crates/ecstore/src/erasure/coding/mod.rs index 21d283c42..ef60a8a1f 100644 --- a/crates/ecstore/src/erasure/coding/mod.rs +++ b/crates/ecstore/src/erasure/coding/mod.rs @@ -20,4 +20,4 @@ pub mod erasure; pub mod heal; pub use bitrot::*; -pub use erasure::{Erasure, ReedSolomonEncoder, calc_shard_size, calc_shard_size_legacy}; +pub use erasure::{Erasure, ErasureConstructionError, ReedSolomonEncoder, calc_shard_size, calc_shard_size_legacy}; diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index 6dbe8e02c..f23e1b863 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -215,11 +215,17 @@ pub enum StorageError { #[error("method not allowed")] MethodNotAllowed, #[error("Io error: {0}")] - Io(std::io::Error), + Io(#[source] std::io::Error), #[error("Lock error: {0}")] Lock(#[from] rustfs_lock::LockError), } +impl From for StorageError { + fn from(error: crate::erasure::coding::ErasureConstructionError) -> Self { + Self::Io(error.into_io_error()) + } +} + impl StorageError { pub fn other(error: E) -> Self where @@ -1176,6 +1182,26 @@ mod tests { use super::*; use std::io::{Error as IoError, ErrorKind}; + #[test] + fn other_preserves_erasure_construction_source_chain() { + use crate::erasure::coding::ErasureConstructionError; + use std::error::Error as _; + + let error = StorageError::from(ErasureConstructionError::ModernEncoder { + source: reed_solomon_erasure::Error::TooManyShards, + }); + let io_source = error.source().expect("StorageError::Io must expose its io::Error source"); + assert!(io_source.is::()); + let construction_source = io_source + .source() + .expect("io::Error must expose the erasure construction error"); + assert!(construction_source.is::()); + let encoder_source = construction_source + .source() + .expect("construction error must expose the encoder error"); + assert!(encoder_source.is::()); + } + // Regression for #952 (ECA-11): an all-`DiskNotFound` slice (every drive in // every set unreachable) must NOT be classified as "all not found", // otherwise ListObjects silently returns an empty listing and masks a full diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 64e988767..a63a92f4d 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,25 @@ pub(crate) fn storage_class_config() -> Option { get_global_storage_class() } +pub(crate) fn storage_class_config_snapshot() -> Arc { + get_global_storage_class_snapshot() +} + +/// Scalar STANDARD / RRS parity for backend-info reporting. +/// +/// Retained for the rebalance/backend-info path. `get_parity_for_sc` returns +/// `None` when the runtime config is uninitialized or (post per-pool support) +/// when pools disagree, so STANDARD falls back to the caller's default and RRS +/// stays `None` — matching the pre-per-pool scalar reporting. +pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option, Option) { + let sc = get_global_storage_class_snapshot(); + 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) +} + pub(crate) fn set_storage_class_config(config: storageclass::Config) { set_global_storage_class(config); } diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index d10b145b6..4df265558 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -163,12 +163,13 @@ impl SetDisks { let erasure = if !latest_meta.deleted && !latest_meta.is_remote() { // Initialize erasure coding; use legacy mode for old-version files - coding::Erasure::new_with_options( + coding::Erasure::try_new_with_options( latest_meta.erasure.data_blocks, latest_meta.erasure.parity_blocks, latest_meta.erasure.block_size, latest_meta.uses_legacy_checksum, ) + .map_err(DiskError::from)? } else { coding::Erasure::default() }; diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 9d4d2d1d9..52612d149 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -355,7 +355,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}")); let result: Result = async { - let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); + let erasure = coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size) + .map_err(Error::from)?; let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); let mut writers = Vec::with_capacity(shuffle_disks.len()); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 5d4618685..d0d912250 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -25,6 +25,11 @@ use crate::bucket::lifecycle::tier_sweeper::delete_object_from_remote_tier_idemp use crate::disk::OldCurrentSize; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; +fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result { + coding::Erasure::try_new_with_options(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size, uses_legacy) + .map_err(Error::from) +} + /// Length of the full plaintext body when — and only when — this read's output /// is exactly the object's complete plaintext, so the app-layer body cache may /// serve it in place of the erasure read. @@ -292,12 +297,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let erasure = coding::Erasure::new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ); + let erasure = erasure_from_file_info(&fi, fi.uses_legacy_checksum)?; let read_length = erasure.shard_file_offset(0, object_size, object_size); let total_shards = data_shards + fi.erasure.parity_blocks; let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files, &fi); @@ -761,7 +761,7 @@ impl SetDisks { let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap()); let result: Result<(ObjectInfo, Option)> = async { - let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); + let erasure = erasure_from_file_info(&fi, false)?; let put_object_size = known_put_object_storage_size(data.size()); let is_inline_buffer = @@ -2886,6 +2886,30 @@ fn drop_failed_writer_disks(disks: &mut [Option], writers: &[Option] committed } +#[cfg(test)] +mod erasure_construction_tests { + use super::*; + use crate::erasure::coding::ErasureConstructionError; + use std::error::Error as _; + + #[test] + fn object_file_info_mapping_preserves_construction_error() { + let mut fi = FileInfo::new("object", 2, 2); + fi.erasure.block_size = 0; + + let error = match erasure_from_file_info(&fi, false) { + Ok(_) => panic!("invalid object erasure metadata must be rejected"), + Err(error) => error, + }; + assert!(error.to_string().contains("block_size must be greater than zero")); + let io_source = error.source().expect("StorageError::Io must expose its io::Error source"); + let construction_source = io_source + .source() + .expect("io::Error must expose the erasure construction error"); + assert!(construction_source.is::()); + } +} + #[cfg(test)] mod b3_write_quorum_tests { use super::drop_failed_writer_disks; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 610fc6136..0ac667c68 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -449,15 +449,13 @@ impl SetDisks { return Ok(None); } - let erasure = coding::Erasure::new_with_options( + let erasure = coding::Erasure::try_new_with_options( fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size, fi.uses_legacy_checksum, - ); - if erasure.data_shards == 0 { - return Ok(None); - } + ) + .map_err(Error::from)?; let checksum_info = fi.erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { @@ -679,21 +677,13 @@ impl SetDisks { object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds" ); - let erasure = coding::Erasure::new_with_options( + let erasure = coding::Erasure::try_new_with_options( fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size, fi.uses_legacy_checksum, - ); - - // Erasure params come from on-disk metadata; zero values must fail the read - // instead of panicking on the block/shard divisions below. - if !erasure.has_valid_dimensions() { - return Err(Error::other(format!( - "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}", - erasure.block_size, erasure.data_shards - ))); - } + ) + .map_err(Error::from)?; let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); @@ -1122,22 +1112,13 @@ impl SetDisks { metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, ) -> Result { - let erasure = coding::Erasure::new_with_options( + let erasure = coding::Erasure::try_new_with_options( fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size, fi.uses_legacy_checksum, - ); - - // Erasure params come from on-disk metadata; zero values must fail the read - // instead of panicking on the block/shard divisions inside the codec - // streaming reader below. Mirrors the legacy multipart guard. - if !erasure.has_valid_dimensions() { - return Err(Error::other(format!( - "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}", - erasure.block_size, erasure.data_shards - ))); - } + ) + .map_err(Error::from)?; let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); @@ -2052,7 +2033,12 @@ mod metadata_cache_tests { ) .await .expect_err("invalid erasure metadata must fail before reader setup"); - assert!(err.to_string().contains("invalid erasure metadata")); + assert!(err.to_string().contains("block_size must be greater than zero")); + let io_source = std::error::Error::source(&err).expect("StorageError::Io must expose its io::Error source"); + let construction_source = io_source + .source() + .expect("io::Error must expose the erasure construction error"); + assert!(construction_source.is::()); assert!(output.is_empty()); } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index bff25424e..14ca9d754 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); @@ -494,9 +519,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}, @@ -508,7 +533,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, @@ -785,33 +812,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()) @@ -836,9 +922,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), @@ -874,16 +962,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..d25fb794b 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,10 +347,19 @@ 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 crate::storage_api::startup::storage::{InstanceContext, bootstrap_instance_ctx}; + use super::{ + global_config_retry_exhausted, init_embedded_startup_storage_foundation, init_startup_storage_foundation, + storage_pool_has_host_failure_risk, + }; + use crate::storage_api::startup::storage::{ + INLINE_BLOCK_ENV, InstanceContext, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV, bootstrap_instance_ctx, + }; use std::sync::Arc; + // The erasure-set-drive-count env var name, kept local to this test to avoid + // reaching across the storage facade boundary for a bare string constant. + const ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT: &str = "RUSTFS_ERASURE_SET_DRIVE_COUNT"; + #[test] fn reports_host_failure_risk_only_for_multi_drive_sets() { assert!(!storage_pool_has_host_failure_risk(0)); @@ -349,6 +373,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 +407,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)); diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 478bd0da3..99192a27f 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -225,6 +225,10 @@ pub(crate) mod startup { } pub(crate) mod storage { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::{ + INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV, + }; pub(crate) use crate::storage::storage_api::{ 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, diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 2980a2e1f..ed9fa9e71 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -140,6 +140,7 @@ ECSTORE_STORAGE_API_ROOT_REEXPORT_HITS_FILE="${TMP_DIR}/ecstore_storage_api_root ECSTORE_STORAGE_API_ROOT_CONSUMER_HITS_FILE="${TMP_DIR}/ecstore_storage_api_root_consumer_hits.txt" ECSTORE_TEST_DIRECT_API_HITS_FILE="${TMP_DIR}/ecstore_test_direct_api_hits.txt" ECSTORE_BENCH_DIRECT_API_HITS_FILE="${TMP_DIR}/ecstore_bench_direct_api_hits.txt" +ERASURE_PANICKING_CONSTRUCTOR_HITS_FILE="${TMP_DIR}/erasure_panicking_constructor_hits.txt" LOCAL_STORAGE_API_RAW_CONTRACT_PATH_HITS_FILE="${TMP_DIR}/local_storage_api_raw_contract_path_hits.txt" STORE_API_DELETE_DTO_REEXPORTS_FILE="${TMP_DIR}/store_api_delete_dto_reexports.txt" STORE_API_DELETE_DTO_INTERNAL_HITS_FILE="${TMP_DIR}/store_api_delete_dto_internal_hits.txt" @@ -1151,6 +1152,54 @@ if [[ -s "$ECSTORE_BENCH_DIRECT_API_HITS_FILE" ]]; then report_failure "ECStore benches must route ECStore and storage-api symbols through crates/ecstore/benches/storage_api/mod.rs: $(paste -sd '; ' "$ECSTORE_BENCH_DIRECT_API_HITS_FILE")" fi +( + cd "$ROOT_DIR" + find rustfs/src crates -type f -name '*.rs' -print0 | + xargs -0 perl -ne ' + next unless $ARGV =~ m{^(?:rustfs/src/|crates/[^/]+/src/)}; + if (!defined $current_file || $ARGV ne $current_file) { + $current_file = $ARGV; + $line = 0; + $pending_test_item = 0; + $in_test_item = 0; + $test_indent = ""; + } + $line++; + if ($in_test_item) { + $in_test_item = 0 if /^(\s*)}\s*[\]),;]*\s*(?:\/\/.*)?$/ && $1 eq $test_indent; + next; + } + if (/^(\s*)#\s*\[\s*cfg\s*\(\s*test\s*\)\s*\]\s*(.*)$/) { + $test_indent = $1; + my $item = $2; + if ($item =~ /^\s*(?:\/\/.*)?$/) { + $pending_test_item = 1; + } elsif ($item !~ /;\s*(?:\/\/.*)?$/ && $item =~ /\{/ && $item !~ /}\s*(?:\/\/.*)?$/) { + $in_test_item = 1; + } elsif ($item !~ /;\s*(?:\/\/.*)?$/ && $item !~ /\{/) { + $pending_test_item = 1; + } + next; + } + if ($pending_test_item) { + if (/;\s*(?:\/\/.*)?$/) { + $pending_test_item = 0; + } elsif (/\{/) { + $pending_test_item = 0; + $in_test_item = 1 unless /}\s*(?:\/\/.*)?$/; + } + next; + } + if (/^\s*(?!\/\/).*\bErasure::new(?:_with_options)?\s*\(/) { + print "$ARGV:$line:$_"; + } + ' || true +) >"$ERASURE_PANICKING_CONSTRUCTOR_HITS_FILE" + +if [[ -s "$ERASURE_PANICKING_CONSTRUCTOR_HITS_FILE" ]]; then + report_failure "production code must use fallible Erasure constructors: $(paste -sd '; ' "$ERASURE_PANICKING_CONSTRUCTOR_HITS_FILE")" +fi + ( cd "$ROOT_DIR" find crates/ecstore/src -type f -name '*.rs' -print0 |