fix(ecstore): resolve erasure parity per pool (#4801) (#5015)

* fix(ecstore): add fallible erasure construction

(cherry picked from commit bd148b20f7)

* fix(ecstore): resolve storage parity per pool

(cherry picked from commit c05c2cb24b)

* 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 <cxymds@gmail.com>

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<usize> {
+    endpoint_pools.as_ref().iter().map(|pool| pool.drives_per_set).collect()
+}
+
+fn resolve_startup_pool_defaults(endpoint_pools: &EndpointServerPools) -> Result<Vec<usize>> {
+    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<Vec<usize>> {
+    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<bool> {
     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<ECStore>, 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<Arc<Self>> {
@@ -186,6 +210,11 @@ impl ECStore {
     ) -> Result<Arc<Self>> {
         // 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::<Vec<_>>(),
+        )
+    }
+
+    #[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<F: Future>(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<crate::runtime::instance::InstanceContext>, Arc<crate::store::ECStore>) {
-        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.

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<DiskStore>, format: &Option<FormatV3
 }
 
 pub fn ec_drives_no_config(set_drive_count: usize) -> Result<usize> {
-    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)]