diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 2c03f3345..4d5d18fba 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -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, + canonical: Vec, + 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) -> 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::(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::(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(pool: Arc, 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) -> Result { + 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, Vec, 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(pools: Vec>, no_lock: bool) -> Result +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, _pools: Vec>) -> 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, pools: Vec>) -> 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(&mut self, pool: Arc) -> 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(&mut self, pools: Vec>) -> Result where S: EcstoreObjectIO, { - self.load_no_lock(pool).await - } - - pub(crate) async fn load_no_lock(&mut self, pool: Arc) -> 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> { @@ -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 { + 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 { + let mut data = Vec::new(); + data.write_u16::(POOL_META_FORMAT) + .expect("pool metadata format should encode"); + data.write_u16::(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, + generation: u64, + } + + let mut data = Vec::new(); + data.write_u16::(POOL_META_FORMAT) + .expect("pool metadata format should encode"); + data.write_u16::(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()); diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index a7aed758d..562cf6807 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -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 diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 322348778..2f1c83ddc 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -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(pool: Arc) -> Result +async fn load_pool_meta_for_startup(pools: Vec>) -> 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(meta: &PoolMeta, pools: Vec>) -> 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( + meta: &PoolMeta, + pools: Vec>, + 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, rx: CancellationToken, pool_indices: Vec) { 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, 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 { + let mut data = Vec::new(); + data.write_u16::(POOL_META_FORMAT) + .expect("pool metadata format should encode"); + data.write_u16::(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, + read_error: bool, read_without_lock: AtomicBool, wrote_without_lock: AtomicBool, wrote_with_max_parity: AtomicBool, + written_payload: Mutex>>, } impl StartupPoolMetaStorage { fn new(read_payload: Vec) -> 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 { 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 { 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");