fix(ecstore): recover pool metadata from replicas (#6457)

2026-08-25 05:26:50 +00:00 · 2026-08-23 23:22:03 +08:00
parent 8ccf7151f3
commit 7f32c675ac
3 changed files with 548 additions and 64 deletions
@@ -30,8 +30,8 @@ use crate::bucket::{
 };
 use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
 use crate::config::com::{
-    CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty,
-    read_config_limited_preserve_empty_with_metadata, read_config_no_lock, save_config, save_config_with_opts,
+    CONFIG_PREFIX, delete_config, read_config_limited_preserve_empty, read_config_limited_preserve_empty_with_metadata,
+    read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty, save_config, save_config_with_opts,
 };
 use crate::data_movement;
 use crate::data_movement::backpressure::{self, DataMovementOperation};
@@ -58,7 +58,11 @@ use crate::storage_api_contracts::{
 };
 use crate::{core::sets::Sets, store::ECStore};
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
-use futures::{StreamExt, future::BoxFuture, stream::FuturesUnordered};
+use futures::{
+    StreamExt,
+    future::{BoxFuture, join_all},
+    stream::FuturesUnordered,
+};
 use http::HeaderMap;
 #[cfg(test)]
 use rmp_serde::Deserializer;
@@ -2152,6 +2156,218 @@ pub struct PoolMeta {
    pub dont_save: bool,
 }

+#[derive(Debug)]
+enum PoolMetaReplica {
+    Missing,
+    Valid {
+        raw: Vec<u8>,
+        canonical: Vec<u8>,
+        meta: PoolMeta,
+    },
+    Corrupt(String),
+    Incompatible(String),
+    Unreadable(String),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct PoolMetaReplicaState {
+    pub(crate) needs_repair: bool,
+    pub(crate) repair_write_safe: bool,
+}
+
+impl PoolMetaReplicaState {
+    pub(crate) fn ensure_write_safe(self, operation: &str) -> Result<()> {
+        if self.repair_write_safe {
+            return Ok(());
+        }
+        Err(Error::other(format!(
+            "{operation}: pool metadata update cannot overwrite an unreadable replica"
+        )))
+    }
+}
+
+#[derive(Debug)]
+struct PoolMetaSelection {
+    meta: PoolMeta,
+    replica_state: PoolMetaReplicaState,
+}
+
+fn classify_pool_meta_tuple_decode_error(kind: &str, err: rmp_serde::decode::Error) -> PoolMetaReplica {
+    let truncated = matches!(
+        &err,
+        rmp_serde::decode::Error::InvalidMarkerRead(source)
+            | rmp_serde::decode::Error::InvalidDataRead(source)
+            if source.kind() == std::io::ErrorKind::UnexpectedEof
+    );
+    if truncated {
+        PoolMetaReplica::Corrupt(format!("{kind} tuple payload is truncated: {err}"))
+    } else {
+        PoolMetaReplica::Incompatible(format!("{kind} tuple payload is not decodable: {err}"))
+    }
+}
+
+fn decode_pool_meta_replica(data: Vec<u8>) -> PoolMetaReplica {
+    if data.len() <= 4 {
+        return PoolMetaReplica::Corrupt("metadata payload is empty or truncated".to_string());
+    }
+
+    let format = LittleEndian::read_u16(&data[0..2]);
+    if format != POOL_META_FORMAT {
+        return PoolMetaReplica::Incompatible(format!("unsupported format {format}"));
+    }
+    let version = LittleEndian::read_u16(&data[2..4]);
+    if version != POOL_META_VERSION {
+        return PoolMetaReplica::Incompatible(format!("unsupported version {version}"));
+    }
+
+    let payload = &data[4..];
+    let meta = match rmp::decode::read_array_len(&mut &payload[..]) {
+        Ok(2) => match rmp_serde::from_slice::<PersistedPoolMeta>(payload) {
+            Ok(meta) => match PoolMeta::try_from(meta) {
+                Ok(meta) => meta,
+                Err(err) => return PoolMetaReplica::Corrupt(err.to_string()),
+            },
+            Err(err) => return classify_pool_meta_tuple_decode_error("current", err),
+        },
+        // V1's third tuple field is the legacy `dont_save` flag. A same-version
+        // boolean extension is byte-identical, so schema extensions must bump
+        // POOL_META_VERSION instead of reusing this shape.
+        Ok(3) => match rmp_serde::from_slice::<LegacyPoolMeta>(payload) {
+            Ok(meta) => match PoolMeta::try_from(meta) {
+                Ok(meta) => meta,
+                Err(err) => return PoolMetaReplica::Corrupt(err.to_string()),
+            },
+            Err(err) => return classify_pool_meta_tuple_decode_error("legacy", err),
+        },
+        Ok(field_count) if field_count < 2 => {
+            return PoolMetaReplica::Corrupt(format!("pool metadata tuple has only {field_count} fields"));
+        }
+        Ok(field_count) => {
+            return PoolMetaReplica::Incompatible(format!("pool metadata tuple has unsupported field count {field_count}"));
+        }
+        Err(_) => {
+            let mut meta = PoolMeta::default();
+            if let Err(err) = meta.load_from_config_data(data.clone()) {
+                let reason = err.to_string();
+                if reason.contains("unknown field") {
+                    return PoolMetaReplica::Incompatible(format!("current-version payload uses unsupported fields: {reason}"));
+                }
+                return PoolMetaReplica::Corrupt(reason);
+            }
+            meta
+        }
+    };
+
+    match meta.encode_config_data() {
+        Ok(canonical) => PoolMetaReplica::Valid {
+            raw: data,
+            canonical,
+            meta,
+        },
+        Err(err) => PoolMetaReplica::Corrupt(err.to_string()),
+    }
+}
+
+async fn read_pool_meta_replica<S>(pool: Arc<S>, no_lock: bool) -> PoolMetaReplica
+where
+    S: EcstoreObjectIO,
+{
+    let result = if no_lock {
+        read_config_no_lock_preserve_empty_with_metadata(pool, POOL_META_NAME)
+            .await
+            .map(|(data, _)| data)
+    } else {
+        read_config_preserve_empty(pool, POOL_META_NAME).await
+    };
+    match result {
+        Ok(data) => decode_pool_meta_replica(data),
+        Err(Error::ConfigNotFound) => PoolMetaReplica::Missing,
+        Err(err) => PoolMetaReplica::Unreadable(err.to_string()),
+    }
+}
+
+fn select_pool_meta_replica(replicas: Vec<PoolMetaReplica>) -> Result<PoolMetaSelection> {
+    if replicas.is_empty() {
+        return Err(Error::other("pool metadata recovery required: no storage pools available"));
+    }
+
+    // V1 has no durable generation. Semantically equivalent legacy/current
+    // encodings can be normalized, but different canonical snapshots require
+    // an operator-selected recovery source instead of an inferred winner.
+    let mut selected: Option<(usize, Vec<u8>, Vec<u8>, PoolMeta)> = None;
+    let mut needs_repair = false;
+    let mut repair_write_safe = true;
+    let mut missing = 0usize;
+    let mut unusable = Vec::new();
+
+    for (idx, replica) in replicas.into_iter().enumerate() {
+        match replica {
+            PoolMetaReplica::Missing => {
+                missing += 1;
+                needs_repair = true;
+            }
+            PoolMetaReplica::Corrupt(reason) => {
+                needs_repair = true;
+                unusable.push(format!("pool {idx} is corrupt: {reason}"));
+            }
+            PoolMetaReplica::Unreadable(reason) => {
+                needs_repair = true;
+                repair_write_safe = false;
+                unusable.push(format!("pool {idx} is unreadable: {reason}"));
+            }
+            PoolMetaReplica::Incompatible(reason) => {
+                return Err(Error::other(format!(
+                    "pool metadata recovery required: pool {idx} is incompatible ({reason}); upgrade or restore a compatible replica without overwriting it"
+                )));
+            }
+            PoolMetaReplica::Valid { raw, canonical, meta } => {
+                if let Some((selected_idx, selected_raw, selected_canonical, _)) = selected.as_ref() {
+                    if selected_canonical != &canonical {
+                        return Err(Error::other(format!(
+                            "pool metadata recovery required: valid replicas in pools {selected_idx} and {idx} diverge; restore one matching pool.bin snapshot before restart"
+                        )));
+                    }
+                    needs_repair |= selected_raw != &raw;
+                } else {
+                    selected = Some((idx, raw, canonical, meta));
+                }
+            }
+        }
+    }
+
+    if let Some((_, _, _, meta)) = selected {
+        return Ok(PoolMetaSelection {
+            meta,
+            replica_state: PoolMetaReplicaState {
+                needs_repair,
+                repair_write_safe,
+            },
+        });
+    }
+    if missing > 0 && unusable.is_empty() {
+        return Ok(PoolMetaSelection {
+            meta: PoolMeta::default(),
+            replica_state: PoolMetaReplicaState {
+                needs_repair: false,
+                repair_write_safe: true,
+            },
+        });
+    }
+
+    Err(Error::other(format!(
+        "pool metadata recovery required: no valid replica is available ({})",
+        unusable.join("; ")
+    )))
+}
+
+async fn load_pool_meta_replicas<S>(pools: Vec<Arc<S>>, no_lock: bool) -> Result<PoolMetaSelection>
+where
+    S: EcstoreObjectIO,
+{
+    let replicas = join_all(pools.into_iter().map(|pool| read_pool_meta_replica(pool, no_lock))).await;
+    select_pool_meta_replica(replicas)
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct PersistedPoolMeta {
@@ -2594,41 +2810,21 @@ impl PoolMeta {
        Ok(())
    }

-    pub async fn load(&mut self, pool: Arc<Sets>, _pools: Vec<Arc<Sets>>) -> Result<()> {
-        let data = match read_config(pool, POOL_META_NAME).await {
-            Ok(data) => data,
-            Err(err) => {
-                if err == Error::ConfigNotFound {
-                    return Ok(());
-                }
-                return Err(err);
-            }
-        };
-        self.load_from_config_data(data)
+    pub async fn load(&mut self, _pool: Arc<Sets>, pools: Vec<Arc<Sets>>) -> Result<()> {
+        let selection = load_pool_meta_replicas(pools, false).await?;
+        *self = selection.meta;
+        Ok(())
    }

-    /// Startup loads pool metadata before the full namespace-lock RPC surface is ready.
-    pub(crate) async fn load_for_startup<S>(&mut self, pool: Arc<S>) -> Result<()>
+    /// Loads every pool metadata replica while the caller owns the metadata fence
+    /// or before the namespace-lock RPC surface is ready during startup.
+    pub(crate) async fn load_no_lock_from_replicas<S>(&mut self, pools: Vec<Arc<S>>) -> Result<PoolMetaReplicaState>
    where
        S: EcstoreObjectIO,
    {
-        self.load_no_lock(pool).await
-    }
-
-    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
-    where
-        S: EcstoreObjectIO,
-    {
-        let data = match read_config_no_lock(pool, POOL_META_NAME).await {
-            Ok(data) => data,
-            Err(err) => {
-                if err == Error::ConfigNotFound {
-                    return Ok(());
-                }
-                return Err(err);
-            }
-        };
-        self.load_from_config_data(data)
+        let selection = load_pool_meta_replicas(pools, true).await?;
+        *self = selection.meta;
+        Ok(selection.replica_state)
    }

    fn encode_config_data(&self) -> Result<Vec<u8>> {
@@ -3502,7 +3698,8 @@ impl ECStore {
            pool_meta.clone()
        };
        let mut latest_pool_meta = PoolMeta::default();
-        latest_pool_meta.load_no_lock(rebalance_pool).await?;
+        let replica_state = latest_pool_meta.load_no_lock_from_replicas(self.pools.clone()).await?;
+        replica_state.ensure_write_safe("decommission start failed")?;
        if latest_pool_meta.pools.is_empty() {
            latest_pool_meta = current_pool_meta;
        }
@@ -7515,6 +7712,167 @@ mod tests {
    use crate::bucket::replication::{ReplicationState, ReplicationStatusType};
    use serde::Serialize;

+    fn pool_meta_replica_test_meta(cmd_line: &str) -> PoolMeta {
+        PoolMeta {
+            version: POOL_META_VERSION,
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: cmd_line.to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: None,
+            }],
+            dont_save: false,
+        }
+    }
+
+    fn pool_meta_replica_test_data(cmd_line: &str) -> Vec<u8> {
+        pool_meta_replica_test_meta(cmd_line)
+            .encode_config_data()
+            .expect("pool metadata should encode")
+    }
+
+    fn pool_meta_legacy_replica_test_data(cmd_line: &str) -> Vec<u8> {
+        let mut data = Vec::new();
+        data.write_u16::<LittleEndian>(POOL_META_FORMAT)
+            .expect("pool metadata format should encode");
+        data.write_u16::<LittleEndian>(POOL_META_VERSION)
+            .expect("pool metadata version should encode");
+        pool_meta_replica_test_meta(cmd_line)
+            .serialize(&mut Serializer::new(&mut data))
+            .expect("legacy pool metadata should encode");
+        data
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_falls_back_from_corrupt_first_copy() {
+        let selection = select_pool_meta_replica(vec![
+            PoolMetaReplica::Corrupt("truncated".to_string()),
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-0")),
+        ])
+        .expect("a validated backup replica should be selected");
+
+        assert!(selection.replica_state.needs_repair);
+        assert!(selection.replica_state.repair_write_safe);
+        assert_eq!(selection.meta.pools.len(), 1);
+        assert_eq!(selection.meta.pools[0].cmd_line, "pool-0");
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_rejects_incompatible_copy() {
+        let valid = pool_meta_replica_test_data("pool-0");
+        let mut incompatible = valid.clone();
+        LittleEndian::write_u16(&mut incompatible[2..4], POOL_META_VERSION + 1);
+
+        let err = select_pool_meta_replica(vec![decode_pool_meta_replica(valid), decode_pool_meta_replica(incompatible)])
+            .expect_err("an incompatible replica must block fallback and repair writes");
+
+        assert!(err.to_string().contains("pool 1 is incompatible"));
+        assert!(err.to_string().contains("without overwriting it"));
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_rejects_partial_write_divergence() {
+        let err = select_pool_meta_replica(vec![
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-old")),
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-new")),
+        ])
+        .expect_err("different valid snapshots have no safe ordering without a generation protocol");
+
+        assert!(err.to_string().contains("valid replicas in pools 0 and 1 diverge"));
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_normalizes_equivalent_legacy_copy() {
+        let selection = select_pool_meta_replica(vec![
+            decode_pool_meta_replica(pool_meta_legacy_replica_test_data("pool-0")),
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-0")),
+        ])
+        .expect("equivalent legacy and current encodings should be compatible");
+
+        assert!(selection.replica_state.needs_repair);
+        assert!(selection.replica_state.repair_write_safe);
+        assert_eq!(selection.meta.pools[0].cmd_line, "pool-0");
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_distinguishes_absent_from_unrecoverable() {
+        assert!(matches!(decode_pool_meta_replica(Vec::new()), PoolMetaReplica::Corrupt(_)));
+
+        let empty = select_pool_meta_replica(vec![PoolMetaReplica::Missing, PoolMetaReplica::Missing])
+            .expect("all missing replicas should preserve new-deployment behavior");
+        assert!(empty.meta.pools.is_empty());
+        assert!(!empty.replica_state.needs_repair);
+        assert!(empty.replica_state.repair_write_safe);
+
+        let err = select_pool_meta_replica(vec![
+            PoolMetaReplica::Missing,
+            PoolMetaReplica::Unreadable("read quorum unavailable".to_string()),
+        ])
+        .expect_err("an unreadable replica must not be treated as a new deployment");
+        assert!(err.to_string().contains("no valid replica is available"));
+        assert!(err.to_string().contains("pool 1 is unreadable"));
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_blocks_repair_for_unreadable_copy() {
+        let selection = select_pool_meta_replica(vec![
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-0")),
+            PoolMetaReplica::Unreadable("read quorum unavailable".to_string()),
+        ])
+        .expect("a validated replica should remain usable while another copy is unreadable");
+
+        assert!(selection.replica_state.needs_repair);
+        assert!(!selection.replica_state.repair_write_safe);
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_rejects_same_version_tuple_extension() {
+        #[derive(Serialize)]
+        struct FuturePersistedPoolMeta {
+            version: u16,
+            pools: Vec<PersistedPoolStatus>,
+            generation: u64,
+        }
+
+        let mut data = Vec::new();
+        data.write_u16::<LittleEndian>(POOL_META_FORMAT)
+            .expect("pool metadata format should encode");
+        data.write_u16::<LittleEndian>(POOL_META_VERSION)
+            .expect("pool metadata version should encode");
+        FuturePersistedPoolMeta {
+            version: POOL_META_VERSION,
+            pools: Vec::new(),
+            generation: 2,
+        }
+        .serialize(&mut Serializer::new(&mut data))
+        .expect("extended tuple pool metadata should encode");
+
+        let err = select_pool_meta_replica(vec![
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-0")),
+            decode_pool_meta_replica(data),
+        ])
+        .expect_err("same-version tuple extensions must block fallback repair writes");
+
+        assert!(err.to_string().contains("pool 1 is incompatible"));
+        assert!(err.to_string().contains("legacy tuple payload is not decodable"));
+    }
+
+    #[test]
+    fn pool_meta_replica_selection_falls_back_from_truncated_current_tuple() {
+        let mut truncated = pool_meta_replica_test_data("pool-truncated");
+        truncated.pop();
+
+        let selection = select_pool_meta_replica(vec![
+            decode_pool_meta_replica(truncated),
+            decode_pool_meta_replica(pool_meta_replica_test_data("pool-valid")),
+        ])
+        .expect("a truncated tuple should not block a validated backup replica");
+
+        assert!(selection.replica_state.needs_repair);
+        assert!(selection.replica_state.repair_write_safe);
+        assert_eq!(selection.meta.pools[0].cmd_line, "pool-valid");
+    }
+
    #[test]
    fn ensure_pool_not_left_in_cmdline_after_decommission_allows_active_pool() {
        assert!(ensure_pool_not_left_in_cmdline_after_decommission(0, "http://node{1...4}/disk{1...4}", false).is_ok());
@@ -109,7 +109,8 @@ impl ECStore {
        }

        let mut pool_meta = PoolMeta::default();
-        pool_meta.load_no_lock(metadata_pool.clone()).await?;
+        let replica_state = pool_meta.load_no_lock_from_replicas(self.pools.clone()).await?;
+        replica_state.ensure_write_safe("heal format fence failed")?;
        if pool_meta.pools.len() != self.pools.len()
            || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
                pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
@@ -13,7 +13,7 @@
 // limitations under the License.

 use super::*;
-use crate::core::pools::{local_decommission_queue_prefix, pool_meta_has_active_decommission};
+use crate::core::pools::{PoolMetaReplicaState, local_decommission_queue_prefix, pool_meta_has_active_decommission};
 use crate::error::is_err_decommission_running;
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -120,13 +120,16 @@ fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<()
    result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}")))
 }

-async fn load_pool_meta_for_startup<S>(pool: Arc<S>) -> Result<PoolMeta>
+async fn load_pool_meta_for_startup<S>(pools: Vec<Arc<S>>) -> Result<(PoolMeta, PoolMetaReplicaState)>
 where
    S: EcstoreObjectIO,
 {
    let mut meta = PoolMeta::default();
-    resolve_store_init_stage_result(meta.load_for_startup(pool).await, "load_pool_meta")?;
-    Ok(meta)
+    let replica_state = meta
+        .load_no_lock_from_replicas(pools)
+        .await
+        .map_err(|err| Error::other(format!("store init failed during load_pool_meta: {err}")))?;
+    Ok((meta, replica_state))
 }

 async fn save_validated_pool_meta_for_startup<S>(meta: &PoolMeta, pools: Vec<Arc<S>>) -> Result<()>
@@ -136,6 +139,28 @@ where
    resolve_store_init_stage_result(meta.save_for_startup(pools).await, "save_validated_pool_meta")
 }

+async fn persist_pool_meta_for_startup_if_safe<S>(
+    meta: &PoolMeta,
+    pools: Vec<Arc<S>>,
+    replica_state: PoolMetaReplicaState,
+    topology_update: bool,
+    elected_writer: bool,
+) -> Result<()>
+where
+    S: EcstoreObjectIO,
+{
+    if !elected_writer {
+        return Ok(());
+    }
+    if topology_update {
+        replica_state.ensure_write_safe("store init failed during save_validated_pool_meta")?;
+    }
+    if topology_update || (replica_state.needs_repair && replica_state.repair_write_safe) {
+        save_validated_pool_meta_for_startup(meta, pools).await?;
+    }
+    Ok(())
+}
+
 async fn resume_local_decommission_after_init(store: Arc<ECStore>, rx: CancellationToken, pool_indices: Vec<usize>) {
    for attempt in 0..=LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES {
        if rx.is_cancelled() {
@@ -450,28 +475,26 @@ impl ECStore {
    pub async fn init(self: &Arc<Self>, rx: CancellationToken) -> Result<()> {
        runtime_sources::ensure_boot_time().await;

-        let meta = load_pool_meta_for_startup(
-            self.pools
-                .first()
-                .cloned()
-                .ok_or_else(|| Error::other("store init failed: no storage pools available"))?,
-        )
-        .await?;
+        let (meta, pool_meta_replica_state) = load_pool_meta_for_startup(self.pools.clone()).await?;
        let update = meta.validate(self.pools.clone())?;
        let endpoints = runtime_sources::endpoint_pools_or_default();
        let should_persist_pool_meta = runtime_sources::first_cluster_node_is_local().await;

-        let installed_pool_meta = if !update {
-            meta.clone()
+        let installed_pool_meta = if update {
+            PoolMeta::new(&self.pools, &meta)
        } else {
-            let new_meta = PoolMeta::new(&self.pools, &meta);
-            // Only one local node should persist validated pool metadata here; otherwise
-            // distributed startup can race on the same lock and replay the prior init bug.
-            if should_persist_pool_meta {
-                save_validated_pool_meta_for_startup(&new_meta, self.pools.clone()).await?;
-            }
-            new_meta
+            meta.clone()
        };
+        // Only one local node should persist validated pool metadata here; otherwise
+        // distributed startup can race on the same lock and replay the prior init bug.
+        persist_pool_meta_for_startup_if_safe(
+            &installed_pool_meta,
+            self.pools.clone(),
+            pool_meta_replica_state,
+            update,
+            should_persist_pool_meta,
+        )
+        .await?;

        {
            let mut pool_meta = self.pool_meta.write().await;
@@ -552,10 +575,11 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
    use super::{
-        LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local,
-        pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with,
-        resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
-        should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
+        LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, persist_pool_meta_for_startup_if_safe,
+        pool_first_endpoint_is_local, pool_meta_has_active_decommission, preflight_startup_rpc_secret_with,
+        resolve_startup_pool_defaults_with, resolve_store_init_stage_result, save_validated_pool_meta_for_startup,
+        should_auto_start_rebalance_after_init, should_retry_format_load, should_retry_local_decommission_resume,
+        wait_for_local_decommission_resume_delay,
    };
    #[cfg(feature = "test-util")]
    use crate::{
@@ -611,7 +635,7 @@ mod tests {
    };
    use crate::{
        bucket::replication::{ReplicationState, ReplicationStatusType, replication_statuses_map},
-        core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus},
+        core::pools::{POOL_META_FORMAT, POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus},
        disk::endpoint::Endpoint,
        error::{Error, Result, StorageError},
        io_support::rio::{WritePlan, compression_metadata_value},
@@ -625,6 +649,7 @@ mod tests {
            range::HTTPRangeSpec,
        },
    };
+    use byteorder::{LittleEndian, WriteBytesExt};
    #[cfg(feature = "test-util")]
    use futures::{StreamExt as _, TryStreamExt as _};
    use http::HeaderMap;
@@ -644,7 +669,7 @@ mod tests {
        future::Future,
        io::Cursor,
        sync::{
-            Arc,
+            Arc, Mutex,
            atomic::{AtomicBool, AtomicUsize, Ordering},
        },
        time::Duration,
@@ -653,21 +678,46 @@ mod tests {
    use tokio::io::AsyncReadExt;
    use tokio_util::sync::CancellationToken;

+    fn startup_pool_meta_payload(meta: &PoolMeta) -> Vec<u8> {
+        let mut data = Vec::new();
+        data.write_u16::<LittleEndian>(POOL_META_FORMAT)
+            .expect("pool metadata format should encode");
+        data.write_u16::<LittleEndian>(POOL_META_VERSION)
+            .expect("pool metadata version should encode");
+        data.extend(rmp_serde::to_vec(meta).expect("legacy pool metadata payload should encode"));
+        data
+    }
+
    #[derive(Debug)]
    struct StartupPoolMetaStorage {
        read_payload: Vec<u8>,
+        read_error: bool,
        read_without_lock: AtomicBool,
        wrote_without_lock: AtomicBool,
        wrote_with_max_parity: AtomicBool,
+        written_payload: Mutex<Option<Vec<u8>>>,
    }

    impl StartupPoolMetaStorage {
        fn new(read_payload: Vec<u8>) -> Self {
            Self {
                read_payload,
+                read_error: false,
                read_without_lock: AtomicBool::new(false),
                wrote_without_lock: AtomicBool::new(false),
                wrote_with_max_parity: AtomicBool::new(false),
+                written_payload: Mutex::new(None),
+            }
+        }
+
+        fn unreadable() -> Self {
+            Self {
+                read_payload: Vec::new(),
+                read_error: true,
+                read_without_lock: AtomicBool::new(false),
+                wrote_without_lock: AtomicBool::new(false),
+                wrote_with_max_parity: AtomicBool::new(false),
+                written_payload: Mutex::new(None),
            }
        }

@@ -702,6 +752,12 @@ mod tests {
        ) -> Result<GetObjectReader> {
            assert!(opts.no_lock, "store init pool metadata load must not require namespace locks");
            self.read_without_lock.store(true, Ordering::SeqCst);
+            if self.read_error {
+                return Err(Error::other("pool metadata read quorum unavailable"));
+            }
+            if self.read_payload.is_empty() {
+                return Err(Error::FileNotFound);
+            }

            Ok(GetObjectReader {
                stream: Box::new(Cursor::new(self.read_payload.clone())),
@@ -715,13 +771,17 @@ mod tests {
            &self,
            bucket: &str,
            object: &str,
-            _data: &mut PutObjReader,
+            data: &mut PutObjReader,
            opts: &ObjectOptions,
        ) -> Result<ObjectInfo> {
            assert!(opts.no_lock, "store init pool metadata save must not require namespace locks");
            self.wrote_without_lock.store(true, Ordering::SeqCst);
            self.wrote_with_max_parity.store(opts.max_parity, Ordering::SeqCst);
-            Ok(self.object_info(bucket, object, 0))
+            let mut payload = Vec::new();
+            data.stream.read_to_end(&mut payload).await?;
+            let size = payload.len();
+            *self.written_payload.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(payload);
+            Ok(self.object_info(bucket, object, size))
        }
    }

@@ -742,10 +802,12 @@ mod tests {
    async fn test_store_init_pool_meta_io_bypasses_namespace_lock_surface() {
        let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));

-        let loaded = load_pool_meta_for_startup(storage.clone())
+        let (loaded, replica_state) = load_pool_meta_for_startup(vec![storage.clone()])
            .await
            .expect("startup pool metadata load should tolerate missing metadata without locks");
        assert!(loaded.pools.is_empty());
+        assert!(!replica_state.needs_repair);
+        assert!(replica_state.repair_write_safe);
        assert!(storage.read_without_lock.load(Ordering::SeqCst));

        let meta = PoolMeta {
@@ -760,6 +822,69 @@ mod tests {
        assert!(storage.wrote_with_max_parity.load(Ordering::SeqCst));
    }

+    #[tokio::test]
+    async fn test_store_init_pool_meta_falls_back_from_corrupt_first_replica() {
+        let corrupt = Arc::new(StartupPoolMetaStorage::new(vec![0, 1, 2]));
+        let expected = init_test_pool_meta(None);
+        let backup = Arc::new(StartupPoolMetaStorage::new(startup_pool_meta_payload(&expected)));
+
+        let (loaded, replica_state) = load_pool_meta_for_startup(vec![corrupt.clone(), backup.clone()])
+            .await
+            .expect("startup should select the validated backup replica");
+
+        assert!(replica_state.needs_repair);
+        assert!(replica_state.repair_write_safe);
+        assert_eq!(loaded.pools.len(), 1);
+        assert_eq!(loaded.pools[0].cmd_line, expected.pools[0].cmd_line);
+        assert!(corrupt.read_without_lock.load(Ordering::SeqCst));
+        assert!(backup.read_without_lock.load(Ordering::SeqCst));
+
+        persist_pool_meta_for_startup_if_safe(&loaded, vec![corrupt.clone(), backup.clone()], replica_state, false, true)
+            .await
+            .expect("the elected startup writer should repair validated corrupt replicas");
+
+        let corrupt_write = corrupt
+            .written_payload
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone()
+            .expect("corrupt replica should be repaired");
+        let backup_write = backup
+            .written_payload
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone()
+            .expect("backup replica should receive the same canonical snapshot");
+        assert_eq!(corrupt_write, backup_write);
+        assert_ne!(corrupt_write, backup.read_payload);
+    }
+
+    #[tokio::test]
+    async fn test_store_init_pool_meta_does_not_repair_unreadable_replica() {
+        let valid = Arc::new(StartupPoolMetaStorage::new(startup_pool_meta_payload(&init_test_pool_meta(None))));
+        let unreadable = Arc::new(StartupPoolMetaStorage::unreadable());
+
+        let (loaded, replica_state) = load_pool_meta_for_startup(vec![valid.clone(), unreadable.clone()])
+            .await
+            .expect("startup should use a validated replica without overwriting an unreadable copy");
+        assert!(replica_state.needs_repair);
+        assert!(!replica_state.repair_write_safe);
+
+        persist_pool_meta_for_startup_if_safe(&loaded, vec![valid.clone(), unreadable.clone()], replica_state, false, true)
+            .await
+            .expect("an unreadable copy should defer repair when no topology write is needed");
+        assert!(!valid.wrote_without_lock.load(Ordering::SeqCst));
+        assert!(!unreadable.wrote_without_lock.load(Ordering::SeqCst));
+
+        let err =
+            persist_pool_meta_for_startup_if_safe(&loaded, vec![valid.clone(), unreadable.clone()], replica_state, true, true)
+                .await
+                .expect_err("a topology update must not overwrite an unreadable replica");
+        assert!(err.to_string().contains("cannot overwrite an unreadable replica"));
+        assert!(!valid.wrote_without_lock.load(Ordering::SeqCst));
+        assert!(!unreadable.wrote_without_lock.load(Ordering::SeqCst));
+    }
+
    #[test]
    fn test_pool_first_endpoint_is_local_respects_local_flag() {
        let mut local_endpoint = Endpoint::try_from("http://127.0.0.1:9000/data").expect("endpoint should parse");