fix: reject unsupported pool expansion with actionable errors (#7360)

* fix: reject unsupported pool expansion with actionable errors

Report singleton-pool and persisted-topology constraints before misleading startup retries. Preserve single-node multi-drive admission and existing parity policies, and cover format preservation plus operator recovery guidance for issue #6186.

* fix: keep pool layout errors typed

Preserve actionable pool layout diagnostics without adding generic formatted errors. Tighten the shrink-only baseline and assert that both typed payloads survive the I/O boundary.

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/config/storageclass.rs

@@ -747,18 +747,27 @@ mod tests {
         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}"
-        );
+        for drives in [2, 3] {
+            let err = lookup_config_for_pools_with_env(&kvs, &[4, drives], no_env_overrides())
+                .expect_err("EC:2 must be rejected by a pool with fewer than four drives per set");
+            assert!(
+                err.to_string().contains("pool 1") && err.to_string().contains(&format!("{drives} drives")),
+                "error must identify the rejecting pool: {err}"
+            );
+        }
+
+        let cfg =
+            lookup_config_for_pools_with_env(&kvs, &[4, 4], no_env_overrides()).expect("EC:2 is valid for both four-drive pools");
+        assert_eq!(cfg.parities_for_sc(STANDARD), Some(vec![2, 2]));
 
         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));
+        for drives in [2, 3, 4] {
+            let cfg =
+                lookup_config_for_pools_with_env(&kvs, &[4, drives], 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, drives), Some(1));
+            assert_eq!(cfg.get_parity_for_sc(STANDARD), Some(1));
+        }
     }
 
     #[test]

crates/ecstore/src/error/mod.rs

@@ -203,6 +203,19 @@ pub enum StorageError {
     NotFirstDisk,
     #[error("first disk wait")]
     FirstDiskWait,
+    #[error(
+        "unsupported pool expansion: an existing single-node single-drive (SNSD) deployment cannot be expanded in place (configured {configured_drives} drive endpoints); restart with the original single local path, or create a new multi-drive deployment and migrate data through S3"
+    )]
+    UnsupportedSnsdExpansion { configured_drives: usize },
+    #[error(
+        "pool topology mismatch: stored {stored_drives} drives with {stored_set_drive_count} drives per erasure set, configured {configured_drives} drives with {configured_set_drive_count} drives per erasure set; an existing pool's drive count and erasure set width cannot be changed in place; restore its original endpoints and RUSTFS_ERASURE_SET_DRIVE_COUNT setting; to expand a multi-drive deployment, append a new pool with at least 2 drive endpoints"
+    )]
+    PoolTopologyMismatch {
+        stored_drives: usize,
+        stored_set_drive_count: usize,
+        configured_drives: usize,
+        configured_set_drive_count: usize,
+    },
 
     // ── Operational ──────────────────────────────────────────────────
     #[error("Storage reached its minimum free drive threshold.")]
@@ -629,6 +642,20 @@ impl Clone for StorageError {
             StorageError::ErasureWriteQuorum => StorageError::ErasureWriteQuorum,
             StorageError::NotFirstDisk => StorageError::NotFirstDisk,
             StorageError::FirstDiskWait => StorageError::FirstDiskWait,
+            StorageError::UnsupportedSnsdExpansion { configured_drives } => StorageError::UnsupportedSnsdExpansion {
+                configured_drives: *configured_drives,
+            },
+            StorageError::PoolTopologyMismatch {
+                stored_drives,
+                stored_set_drive_count,
+                configured_drives,
+                configured_set_drive_count,
+            } => StorageError::PoolTopologyMismatch {
+                stored_drives: *stored_drives,
+                stored_set_drive_count: *stored_set_drive_count,
+                configured_drives: *configured_drives,
+                configured_set_drive_count: *configured_set_drive_count,
+            },
             StorageError::TooManyOpenFiles => StorageError::TooManyOpenFiles,
             StorageError::NoHealRequired => StorageError::NoHealRequired,
             StorageError::Lock(e) => StorageError::Lock(e.clone()),
@@ -735,6 +762,11 @@ impl StorageError {
             StorageError::ErasureWriteQuorum => StorageErrorCode::ErasureWriteQuorum,
             StorageError::NotFirstDisk => StorageErrorCode::NotFirstDisk,
             StorageError::FirstDiskWait => StorageErrorCode::FirstDiskWait,
+            // Topology diagnostics reuse the existing wire code; they are
+            // not disk errors and must retain their local identity for retry classification.
+            StorageError::UnsupportedSnsdExpansion { .. } | StorageError::PoolTopologyMismatch { .. } => {
+                StorageErrorCode::InvalidArgument
+            }
             StorageError::ConfigNotFound => StorageErrorCode::ConfigNotFound,
             StorageError::TooManyOpenFiles => StorageErrorCode::TooManyOpenFiles,
             StorageError::NoHealRequired => StorageErrorCode::NoHealRequired,
@@ -1215,6 +1247,29 @@ mod tests {
     use super::*;
     use std::io::{Error as IoError, ErrorKind};
 
+    #[test]
+    fn startup_topology_errors_preserve_identity_and_guidance() {
+        for error in [
+            StorageError::UnsupportedSnsdExpansion { configured_drives: 4 },
+            StorageError::PoolTopologyMismatch {
+                stored_drives: 4,
+                stored_set_drive_count: 4,
+                configured_drives: 8,
+                configured_set_drive_count: 8,
+            },
+        ] {
+            let io_error: IoError = error.clone().into();
+            let restored = StorageError::from(io_error);
+            assert_eq!(std::mem::discriminant(&restored), std::mem::discriminant(&error));
+            assert_eq!(restored.to_string(), error.to_string());
+            assert_eq!(restored.code(), StorageErrorCode::InvalidArgument);
+            assert!(
+                restored.narrow_to_disk().is_err(),
+                "startup diagnostics must not become disk/quorum errors"
+            );
+        }
+    }
+
     #[test]
     fn other_preserves_erasure_construction_source_chain() {
         use crate::erasure::coding::ErasureConstructionError;

crates/ecstore/src/layout/disks_layout.rs

@@ -25,6 +25,20 @@ pub(crate) const MAX_ERASURE_SET_DRIVE_COUNT: usize = 16;
 const SET_SIZES: [usize; 15] = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, MAX_ERASURE_SET_DRIVE_COUNT];
 const ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT: &str = "RUSTFS_ERASURE_SET_DRIVE_COUNT";
 
+#[derive(Debug, thiserror::Error)]
+enum PoolDriveCountError {
+    #[error(
+        "Incorrect number of endpoints provided, size {size}; an erasure pool requires at least {} drive endpoints on one or more nodes; for a standalone single-drive deployment, use a single local path without ellipses",
+        SET_SIZES[0]
+    )]
+    BelowMinimum { size: usize },
+    #[error(
+        "Incorrect number of endpoints provided, size {size}; {}={set_drive_count} requires at least {set_drive_count} drive endpoints per pool",
+        ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT
+    )]
+    BelowSetWidth { size: usize, set_drive_count: usize },
+}
+
 #[derive(Deserialize, Debug, Default)]
 pub struct PoolDisksLayout {
     cmd_line: String,
@@ -132,7 +146,7 @@ impl DisksLayout {
         for arg in args.iter() {
             if !has_ellipses(&[arg]) && args.len() > 1 {
                 return Err(Error::other(
-                    "all args must have ellipses for pool expansion (Invalid arguments specified)",
+                    "all args must have ellipses for pool expansion (Invalid arguments specified); each pool must expand to at least 2 drive endpoints on one or more nodes; a single-drive pool cannot be added to a multi-pool deployment",
                 ));
             }
 
@@ -396,9 +410,11 @@ fn get_set_indexes<T: AsRef<str>>(
     }
 
     for &size in total_sizes {
-        // Check if total_sizes has minimum range upto set_size
-        if size < SET_SIZES[0] || size < set_drive_count {
-            return Err(Error::other(format!("Incorrect number of endpoints provided, size {size}")));
+        if size < SET_SIZES[0] {
+            return Err(Error::other(PoolDriveCountError::BelowMinimum { size }));
+        }
+        if size < set_drive_count {
+            return Err(Error::other(PoolDriveCountError::BelowSetWidth { size, set_drive_count }));
         }
     }
 
@@ -707,7 +723,7 @@ mod test {
                 arg: "http://rustfs{2...3}/export/set{1...0}",
                 ..Default::default()
             },
-            // Range cannot be smaller than 4 minimum.
+            // Ranges must use three dots.
             TestCase {
                 num: 4,
                 arg: "/export{1..2}",
@@ -926,11 +942,146 @@ mod test {
         }
     }
 
+    #[test]
+    fn pool_expansion_accepts_single_node_multi_drive_pools() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            for (volumes, drives) in [
+                (["http://node1:9000/data{1...2}", "http://node2:9000/data{1...2}"], 2),
+                (["http://node1:9000/data{1...4}", "http://node2:9000/data{1...4}"], 4),
+                (["http://node{1...4}:9000/data", "http://node5:9000/data{1...4}"], 4),
+                (["http://node5:9000/data{1...4}", "http://node{1...4}:9000/data"], 4),
+            ] {
+                let layout = DisksLayout::from_volumes(&volumes).expect("single-node multi-drive pools are valid");
+
+                assert!(!layout.legacy);
+                assert_eq!(layout.pools.len(), 2);
+                for (index, volume) in volumes.iter().enumerate() {
+                    assert_eq!(layout.get_set_count(index), 1);
+                    assert_eq!(layout.get_drives_per_set(index), drives);
+                    assert_eq!(layout.get_cmd_line(index), *volume);
+                }
+            }
+        });
+    }
+
+    #[test]
+    fn pool_expansion_accepts_multi_node_single_drive_pools() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            for nodes in [2, 3, 4] {
+                let volumes = [
+                    format!("http://pool1-node{{1...{nodes}}}:9000/data"),
+                    format!("http://pool2-node{{1...{nodes}}}:9000/data"),
+                ];
+                let layout = DisksLayout::from_volumes(&volumes).expect("each node may contribute one drive to a pool");
+
+                assert_eq!(layout.pools.len(), 2);
+                for pool in 0..2 {
+                    assert_eq!(layout.get_set_count(pool), 1);
+                    assert_eq!(layout.get_drives_per_set(pool), nodes);
+                    let expected = (1..=nodes)
+                        .map(|node| format!("http://pool{}-node{node}:9000/data", pool + 1))
+                        .collect::<Vec<_>>();
+                    assert_eq!(layout.pools[pool].layout, vec![expected]);
+                }
+            }
+        });
+    }
+
+    #[test]
+    fn explicit_endpoints_without_ellipses_form_one_pool() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            let volumes = ["http://node1:9000/data", "http://node2:9000/data"];
+            let layout = DisksLayout::from_volumes(&volumes).expect("explicit endpoints form one legacy pool");
+
+            assert!(layout.legacy);
+            assert_eq!(layout.pools.len(), 1);
+            assert_eq!(layout.pools[0].layout, vec![volumes.to_vec()]);
+        });
+    }
+
+    #[test]
+    fn standalone_single_drive_path_remains_supported() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            let layout = DisksLayout::from_volumes(&["/data"]).expect("standalone single-drive deployment is valid");
+
+            assert!(layout.is_single_drive_layout());
+            assert_eq!(layout.get_single_drive_layout(), "/data");
+        });
+    }
+
+    #[test]
+    fn pool_expansion_rejects_plain_single_drive_pool_with_notice() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            for volumes in [
+                ["http://node{1...2}:9000/data", "http://node3:9000/data"],
+                ["http://node3:9000/data", "http://node{1...2}:9000/data"],
+            ] {
+                let err = DisksLayout::from_volumes(&volumes).expect_err("a plain endpoint cannot be an expansion pool");
+                let message = err.to_string();
+
+                assert!(message.contains("all args must have ellipses for pool expansion"), "{message}");
+                assert!(message.contains("at least 2 drive endpoints"), "{message}");
+            }
+        });
+    }
+
+    #[test]
+    fn pool_expansion_rejects_singleton_ellipsis_pool_with_notice() {
+        temp_env::with_var(ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT, Some("0"), || {
+            for singleton in ["http://node{3...3}:9000/data", "http://node3:9000/data{1...1}"] {
+                for volumes in [
+                    vec!["http://node{1...2}:9000/data", singleton],
+                    vec![singleton, "http://node{1...2}:9000/data"],
+                    vec![singleton],
+                ] {
+                    let err = DisksLayout::from_volumes(&volumes).expect_err("a singleton range still contains one drive");
+                    let message = err.to_string();
+
+                    assert_eq!(err.kind(), std::io::ErrorKind::Other);
+                    assert!(matches!(
+                        err.get_ref().and_then(|source| source.downcast_ref::<PoolDriveCountError>()),
+                        Some(PoolDriveCountError::BelowMinimum { size: 1 })
+                    ));
+                    assert!(message.contains("at least 2 drive endpoints"), "{message}");
+                    assert!(message.contains("single local path without ellipses"), "{message}");
+                }
+            }
+        });
+    }
+
+    #[test]
+    fn explicit_set_size_counts_drives_not_nodes() {
+        for volume in ["http://node1:9000/data{1...4}", "http://node{1...4}:9000/data"] {
+            let sets = get_all_sets(2, true, &[volume]).expect("four endpoints can form two two-drive sets");
+            assert_eq!(sets.iter().map(Vec::len).collect::<Vec<_>>(), vec![2, 2]);
+        }
+    }
+
+    #[test]
+    fn undersized_pool_error_identifies_requested_set_size() {
+        let err =
+            get_all_sets(4, true, &["http://node{1...2}:9000/data"]).expect_err("two endpoints cannot fill a four-drive set");
+        let message = err.to_string();
+
+        assert_eq!(err.kind(), std::io::ErrorKind::Other);
+        assert!(matches!(
+            err.get_ref().and_then(|source| source.downcast_ref::<PoolDriveCountError>()),
+            Some(PoolDriveCountError::BelowSetWidth {
+                size: 2,
+                set_drive_count: 4
+            })
+        ));
+        assert!(message.contains("size 2"), "{message}");
+        assert!(message.contains("RUSTFS_ERASURE_SET_DRIVE_COUNT=4"), "{message}");
+    }
+
     #[test]
     fn layout_errors_do_not_echo_url_credentials() {
         for volumes in [
             vec!["http://:duplicate-secret@server/path", "http://:duplicate-secret@server/path"],
             vec!["http://:ellipsis...secret@server/path"],
+            vec!["http://server{1...2}/data", "http://:plain-secret@server3/data"],
+            vec!["http://server{1...2}/data", "http://:singleton-secret@server{3...3}/data"],
         ] {
             let err = DisksLayout::from_volumes(&volumes).unwrap_err();
             assert!(!err.to_string().contains("secret"), "layout error leaked endpoint credentials: {err}");

crates/ecstore/src/layout/endpoints.rs

@@ -2432,6 +2432,41 @@ mod test {
         assert_eq!(local_endpoints[0].pool_idx, 1);
     }
 
+    #[tokio::test]
+    async fn pool_expansion_resolves_single_node_multi_drive_and_multi_node_single_drive_pools() {
+        for (additional_pool, expected_nodes) in [
+            ("http://rustfs-5.example.invalid:9000/data{1...4}", 5),
+            ("http://rustfs-{5...8}.example.invalid:9000/data", 8),
+        ] {
+            let layout = temp_env::with_var("RUSTFS_ERASURE_SET_DRIVE_COUNT", Some("0"), || {
+                DisksLayout::from_volumes(&["http://rustfs-{1...4}.example.invalid:9000/data", additional_pool])
+            })
+            .expect("both single-node multi-drive and multi-node single-drive pools should parse");
+
+            let (pools, setup_type) = EndpointServerPools::create_server_endpoints_with(
+                "0.0.0.0:9000",
+                &layout,
+                Some(orchestrated_test_policy()),
+                Some("rustfs-1.example.invalid"),
+            )
+            .await
+            .expect("pool admission must not impose a minimum node count or drives per node");
+
+            assert_eq!(setup_type, SetupType::DistErasure);
+            assert_eq!(pools.0.len(), 2);
+            assert_eq!(pools.get_nodes().len(), expected_nodes);
+            for (pool_index, pool) in (0_i32..).zip(&pools.0) {
+                assert_eq!((pool.set_count, pool.drives_per_set), (1, 4));
+                assert_eq!(pool.endpoints.as_ref().len(), 4);
+                for (disk_index, endpoint) in (0_i32..).zip(pool.endpoints.as_ref()) {
+                    assert_eq!(endpoint.pool_idx, pool_index);
+                    assert_eq!(endpoint.set_idx, 0);
+                    assert_eq!(endpoint.disk_idx, disk_index);
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn explicit_local_endpoint_host_fails_closed_for_invalid_context_or_zero_match() {
         let args = vec![

crates/ecstore/src/store/init.rs

@@ -103,7 +103,10 @@ const REBALANCE_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(10);
 const REBALANCE_RESUME_RETRY_DELAY: Duration = Duration::from_secs(10);
 
 fn should_retry_format_load(err: &Error) -> bool {
-    !matches!(err, Error::CorruptedFormat)
+    !matches!(
+        err,
+        Error::CorruptedFormat | Error::UnsupportedSnsdExpansion { .. } | Error::PoolTopologyMismatch { .. }
+    )
 }
 
 fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_resume_required: bool) -> bool {
@@ -1784,6 +1787,33 @@ mod tests {
         assert!(should_retry_format_load(&StorageError::FirstDiskWait));
     }
 
+    #[test]
+    fn test_should_retry_format_load_rejects_permanent_topology_errors() {
+        for error in [
+            StorageError::UnsupportedSnsdExpansion { configured_drives: 4 },
+            StorageError::PoolTopologyMismatch {
+                stored_drives: 4,
+                stored_set_drive_count: 4,
+                configured_drives: 8,
+                configured_set_drive_count: 8,
+            },
+        ] {
+            assert!(!should_retry_format_load(&error), "topology errors require operator action: {error}");
+        }
+        for error in [
+            StorageError::DiskNotFound,
+            StorageError::Timeout,
+            StorageError::RemoteNotInitialized,
+            StorageError::NotFirstDisk,
+            StorageError::other(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)),
+        ] {
+            assert!(
+                should_retry_format_load(&error),
+                "transient failures retain their existing retry path: {error}"
+            );
+        }
+    }
+
     #[test]
     fn test_should_auto_start_rebalance_after_init_allows_active_rebalance_without_decommission() {
         assert!(should_auto_start_rebalance_after_init(false, true));

crates/ecstore/src/store/init_format.rs

@@ -109,6 +109,21 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
     let fresh_bootstrap_proven = should_init_erasure_disks(&errs);
     let formats_present = formats.iter().flatten().count();
     let mut format_quorum = (formats_present > 0).then(|| select_format_erasure_in_quorum(&formats, 0));
+    // A resized pool may never reach quorum under its new endpoint count.
+    // Diagnose a valid, unambiguous stored layout before migration or waiting.
+    // A healthy quorum still takes precedence over foreign minority formats;
+    // conflicting or malformed observations retain their existing error path.
+    if format_quorum.as_ref().is_some_and(Result::is_err)
+        && let Some(reference) = formats.iter().flatten().next()
+        && formats.iter().flatten().all(|format| {
+            format.shared_identity() == reference.shared_identity()
+                && reference.erasure.sets.iter().flatten().any(|id| *id == format.erasure.this)
+        })
+        && let Err(err @ (Error::UnsupportedSnsdExpansion { .. } | Error::PoolTopologyMismatch { .. })) =
+            check_format_erasure_value_for_topology(reference, formats.len(), set_drive_count)
+    {
+        return Err(err);
+    }
     if format_quorum.as_ref().is_none_or(Result::is_err)
         && errs.iter().any(|error| {
             matches!(
@@ -661,15 +676,18 @@ fn check_format_erasure_value_for_topology(format: &FormatV3, format_count: usiz
         .len()
         .checked_mul(set_drive_count_in_format)
         .ok_or_else(|| Error::other("erasure set drive count overflow"))?;
-    if format_count != format_drive_count {
-        return Err(Error::other(format!(
-            "formats length for erasure.sets does not match: got {format_count}, expected {format_drive_count}"
-        )));
+    if format_drive_count == 1 && format_count > 1 {
+        return Err(Error::UnsupportedSnsdExpansion {
+            configured_drives: format_count,
+        });
     }
-    if set_drive_count_in_format != set_drive_count {
-        return Err(Error::other(format!(
-            "erasure set length for set_drive_count does not match: got {set_drive_count_in_format}, expected {set_drive_count}"
-        )));
+    if format_count != format_drive_count || set_drive_count_in_format != set_drive_count {
+        return Err(Error::PoolTopologyMismatch {
+            stored_drives: format_drive_count,
+            stored_set_drive_count: set_drive_count_in_format,
+            configured_drives: format_count,
+            configured_set_drive_count: set_drive_count,
+        });
     }
     Ok(())
 }
@@ -877,6 +895,10 @@ mod tests {
     use serial_test::serial;
 
     async fn local_disks(count: usize) -> (tempfile::TempDir, Vec<Option<DiskStore>>) {
+        local_disks_with_set_width(count, count).await
+    }
+
+    async fn local_disks_with_set_width(count: usize, set_width: usize) -> (tempfile::TempDir, Vec<Option<DiskStore>>) {
         let temp_dir = tempfile::tempdir().expect("temporary disk root should be created");
         let mut endpoints = Vec::with_capacity(count);
         for disk_index in 0..count {
@@ -887,8 +909,8 @@ mod tests {
             let mut endpoint =
                 Endpoint::try_from(path.to_str().expect("temporary disk path should be UTF-8")).expect("endpoint should parse");
             endpoint.set_pool_index(0);
-            endpoint.set_set_index(0);
-            endpoint.set_disk_index(disk_index);
+            endpoint.set_set_index(disk_index / set_width);
+            endpoint.set_disk_index(disk_index % set_width);
             endpoints.push(endpoint);
         }
 
@@ -912,6 +934,21 @@ mod tests {
         (temp_dir, disks)
     }
 
+    async fn format_bytes(disks: &[Option<DiskStore>]) -> Vec<Option<Vec<u8>>> {
+        let mut snapshots = Vec::with_capacity(disks.len());
+        for disk in disks {
+            let disk = disk.as_ref().expect("snapshot disk should exist");
+            // Inspect bytes even when the disk wrapper rejects a format whose
+            // stored slot differs from the attempted new endpoint geometry.
+            match tokio::fs::read(disk.path().join(RUSTFS_META_BUCKET).join(FORMAT_CONFIG_FILE)).await {
+                Ok(data) => snapshots.push(Some(data)),
+                Err(err) if err.kind() == std::io::ErrorKind::NotFound => snapshots.push(None),
+                Err(err) => panic!("format snapshot failed: {err}"),
+            }
+        }
+        snapshots
+    }
+
     async fn write_legacy_format(disk: &Option<DiskStore>, format: &FormatV3) {
         write_legacy_bytes(disk, bytes::Bytes::from(format.to_json().expect("legacy format should serialize"))).await;
     }
@@ -1116,6 +1153,212 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn single_drive_format_rejects_in_place_expansion_without_writes() {
+        for configured_drives in [2, 4] {
+            for first_disk in [false, true] {
+                let (_temp_dir, mut disks) = local_disks(configured_drives).await;
+                let mut original = FormatV3::new(1, 1);
+                original.erasure.this = original.erasure.sets[0][0];
+                save_format_file(&disks[0], &Some(original))
+                    .await
+                    .expect("SNSD format should be written");
+                let before = format_bytes(&disks).await;
+
+                let err = connect_load_init_formats(first_disk, &mut disks, 1, configured_drives, None)
+                    .await
+                    .expect_err("an existing SNSD deployment cannot expand in place");
+                let message = err.to_string();
+                assert!(message.contains("SNSD"), "expected a single-drive expansion error: {message}");
+                assert!(message.contains("migrate data through S3"), "expected actionable guidance: {message}");
+                assert_eq!(format_bytes(&disks).await, before, "neither old nor new formats may be written");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn existing_pool_rejects_drive_count_or_set_width_changes_without_writes() {
+        for (stored_sets, stored_width, configured_sets, configured_width) in
+            [(1, 4, 1, 6), (1, 4, 1, 8), (1, 4, 1, 2), (1, 4, 2, 2), (2, 2, 1, 4)]
+        {
+            for first_disk in [false, true] {
+                let (_temp_dir, mut disks) =
+                    local_disks_with_set_width(configured_sets * configured_width, configured_width).await;
+                let original = FormatV3::new(stored_sets, stored_width);
+                for (disk, disk_id) in disks.iter().zip(original.erasure.sets.iter().flatten()) {
+                    let mut format = original.clone();
+                    format.erasure.this = *disk_id;
+                    save_format_file(disk, &Some(format))
+                        .await
+                        .expect("existing format should be written");
+                }
+                let before = format_bytes(&disks).await;
+
+                let err = connect_load_init_formats(first_disk, &mut disks, configured_sets, configured_width, None)
+                    .await
+                    .expect_err("an existing pool's geometry is immutable");
+                let message = err.to_string();
+                assert!(message.contains("pool topology mismatch"), "expected a topology error: {message}");
+                assert!(
+                    message.contains(&format!("stored 4 drives with {stored_width} drives per erasure set")),
+                    "expected stored geometry: {message}"
+                );
+                assert!(message.contains("append a new pool"), "expected expansion guidance: {message}");
+                assert_eq!(format_bytes(&disks).await, before, "rejection must not rewrite any format");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn subquorum_existing_layout_with_missing_drives_is_not_expansion() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let mut original = FormatV3::new(1, 4);
+        original.erasure.this = original.erasure.sets[0][0];
+        save_format_file(&disks[0], &Some(original))
+            .await
+            .expect("existing format should be written");
+        disks.extend([None, None, None]);
+
+        for first_disk in [false, true] {
+            assert!(matches!(
+                connect_load_init_formats(first_disk, &mut disks, 1, 4, None).await,
+                Err(Error::ErasureReadQuorum)
+            ));
+        }
+    }
+
+    #[tokio::test]
+    async fn conflicting_layouts_without_quorum_are_not_expansion_proof() {
+        let (_temp_dir, mut disks) = local_disks(2).await;
+        for (index, (disk, width)) in disks.iter().zip([4, 2]).enumerate() {
+            let mut format = FormatV3::new(1, width);
+            format.erasure.this = format.erasure.sets[0][index];
+            save_format_file(disk, &Some(format))
+                .await
+                .expect("existing format should be written");
+        }
+        disks.extend([None, None]);
+
+        let result = connect_load_init_formats(true, &mut disks, 1, 4, None).await;
+        assert!(matches!(result, Err(Error::ErasureReadQuorum)), "conflicting layout result: {result:?}");
+    }
+
+    #[tokio::test]
+    async fn existing_format_quorum_ignores_single_drive_outlier() {
+        let (_temp_dir, mut disks) = local_disks(3).await;
+        let majority = FormatV3::new(1, 3);
+        for (index, disk) in disks.iter().enumerate() {
+            // Slot zero lets the SNSD outlier pass the disk wrapper's own
+            // slot check, so quorum selection must exclude the parsed format.
+            let mut format = if index == 0 { FormatV3::new(1, 1) } else { majority.clone() };
+            format.erasure.this = format.erasure.sets[0][index];
+            save_format_file(disk, &Some(format))
+                .await
+                .expect("existing format should be written");
+        }
+
+        let loaded = connect_load_init_formats(true, &mut disks, 1, 3, None)
+            .await
+            .expect("a foreign SNSD outlier must not block a healthy majority");
+        assert_eq!(loaded.shared_identity(), majority.shared_identity());
+        assert!(disks[0].is_none(), "the foreign single-drive format must be quarantined");
+    }
+
+    #[tokio::test]
+    async fn multi_drive_pool_expansion_preserves_existing_format() {
+        let (_original_dir, mut disks) = local_disks(4).await;
+        let (_new_dir, mut new_disks) = local_disks(4).await;
+        let original = connect_load_init_formats(true, &mut disks, 1, 4, None)
+            .await
+            .expect("original multi-drive pool should initialize");
+        let before = format_bytes(&disks).await;
+
+        let added = connect_load_init_formats(true, &mut new_disks, 1, 4, Some(original.id))
+            .await
+            .expect("a new multi-drive pool should initialize with the existing deployment ID");
+        assert_eq!(added.id, original.id);
+        assert_ne!(added.erasure.sets, original.erasure.sets);
+        assert_eq!(format_bytes(&disks).await, before);
+        assert_eq!(
+            connect_load_init_formats(true, &mut disks, 1, 4, Some(original.id))
+                .await
+                .expect("the original pool should restart with unchanged geometry"),
+            original
+        );
+        assert_eq!(
+            connect_load_init_formats(true, &mut new_disks, 1, 4, Some(original.id))
+                .await
+                .expect("the new pool should restart with its own format"),
+            added
+        );
+    }
+
+    #[tokio::test]
+    async fn store_startup_rejects_pool_resize_before_retry_loop() {
+        use crate::layout::endpoints::{EndpointServerPools, PoolEndpoints};
+        use tokio_util::sync::CancellationToken;
+
+        for (stored_width, configured_width) in [(1, 4), (4, 8)] {
+            let (_temp_dir, disks) = local_disks(configured_width).await;
+            let original = FormatV3::new(1, stored_width);
+            for (disk, disk_id) in disks.iter().zip(&original.erasure.sets[0]) {
+                let mut format = original.clone();
+                format.erasure.this = *disk_id;
+                save_format_file(disk, &Some(format))
+                    .await
+                    .expect("old format should be written");
+            }
+            let before = format_bytes(&disks).await;
+            let endpoints = disks.iter().flatten().map(|disk| disk.endpoint()).collect::<Vec<_>>();
+            let pools = EndpointServerPools::from(vec![PoolEndpoints {
+                legacy: true,
+                set_count: 1,
+                drives_per_set: configured_width,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: "test-pool".to_string(),
+                platform: String::new(),
+            }]);
+            let shutdown = CancellationToken::new();
+            let result = temp_env::async_with_vars(
+                [
+                    (storageclass::STANDARD_ENV, None::<&str>),
+                    (storageclass::RRS_ENV, None::<&str>),
+                    (storageclass::OPTIMIZE_ENV, None::<&str>),
+                    (storageclass::INLINE_BLOCK_ENV, None::<&str>),
+                ],
+                tokio::time::timeout(
+                    std::time::Duration::from_secs(5),
+                    crate::store::ECStore::new_with_instance_ctx(
+                        "127.0.0.1:0".parse().expect("test address"),
+                        pools,
+                        shutdown.clone(),
+                        Arc::new(InstanceContext::new()),
+                    ),
+                ),
+            )
+            .await;
+            shutdown.cancel();
+            let err = result
+                .expect("invalid topology must abort without the format retry backoff")
+                .expect_err("resize must fail");
+            match stored_width {
+                1 => assert!(matches!(err, Error::UnsupportedSnsdExpansion { configured_drives: 4 }), "{err}"),
+                _ => assert!(
+                    matches!(
+                        err,
+                        Error::PoolTopologyMismatch {
+                            stored_drives: 4,
+                            configured_drives: 8,
+                            ..
+                        }
+                    ),
+                    "{err}"
+                ),
+            }
+            assert_eq!(format_bytes(&disks).await, before, "failed store startup must not write formats");
+        }
+    }
+
     #[tokio::test]
     async fn existing_format_load_rejects_conflicting_formats_without_a_majority() {
         let (_temp_dir, mut disks) = two_local_disks_with_missing_third().await;