diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 6ce79f29f..42797dda0 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -168,6 +168,19 @@ pub const DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED: bool = false; const _: () = assert!(!DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_WRITE); const _: () = assert!(!DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED); +/// Request writing pool metadata version 2. +/// +/// This remains ineffective until [`ENV_POOL_META_V2_FLEET_CONFIRMED`] is also enabled. +pub const ENV_POOL_META_V2_WRITE: &str = "RUSTFS_POOL_META_V2_WRITE"; +pub const DEFAULT_POOL_META_V2_WRITE: bool = false; + +/// Operator-attested confirmation that every pool metadata reader and writer understands version 2. +pub const ENV_POOL_META_V2_FLEET_CONFIRMED: &str = "RUSTFS_POOL_META_V2_FLEET_CONFIRMED"; +pub const DEFAULT_POOL_META_V2_FLEET_CONFIRMED: bool = false; + +const _: () = assert!(!DEFAULT_POOL_META_V2_WRITE); +const _: () = assert!(!DEFAULT_POOL_META_V2_FLEET_CONFIRMED); + // ============================================================================= // Concurrent Request Fix - Timeout and Backpressure Configuration // ============================================================================= @@ -736,4 +749,10 @@ mod remote_version_state_tests { "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED" ); } + + #[test] + fn pool_meta_v2_gate_uses_stable_environment_names() { + assert_eq!(super::ENV_POOL_META_V2_WRITE, "RUSTFS_POOL_META_V2_WRITE"); + assert_eq!(super::ENV_POOL_META_V2_FLEET_CONFIRMED, "RUSTFS_POOL_META_V2_FLEET_CONFIRMED"); + } } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 54f9996a4..dfd520cb6 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -243,8 +243,8 @@ pub mod cache { pub mod capacity { pub use crate::core::pools::{ - PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free, path2_bucket_object, - path2_bucket_object_with_base_path, + DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free, + path2_bucket_object, path2_bucket_object_with_base_path, }; pub use crate::store::utils::is_reserved_or_invalid_bucket; } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index b1f4723d7..3790ba7b5 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -40,7 +40,7 @@ use crate::data_movement; use crate::data_movement::backpressure::{self, DataMovementOperation}; use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::disk::error::DiskError; -use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; +use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::error::{ StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled, @@ -131,7 +131,22 @@ const DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT: std::time::Duration = std:: pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; -pub const POOL_META_VERSION: u16 = 1; +const POOL_META_V1_VERSION: u16 = 1; +pub const POOL_META_VERSION: u16 = 2; + +fn pool_meta_v2_writer_enabled_for(requested: bool, fleet_confirmed: bool) -> bool { + requested && fleet_confirmed +} + +fn pool_meta_v2_writer_enabled() -> bool { + pool_meta_v2_writer_enabled_for( + rustfs_utils::get_env_bool(rustfs_config::ENV_POOL_META_V2_WRITE, rustfs_config::DEFAULT_POOL_META_V2_WRITE), + rustfs_utils::get_env_bool( + rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, + rustfs_config::DEFAULT_POOL_META_V2_FLEET_CONFIRMED, + ), + ) +} #[derive(Clone, Debug)] pub struct DecommissionCanceler { @@ -567,6 +582,12 @@ fn build_decommission_start_state( info.decommissioned_buckets = previous.decommissioned_buckets.clone(); info.items_decommissioned = previous.items_decommissioned; info.bytes_done = previous.bytes_done; + info.unresolved_entries = previous.unresolved_entries.clone(); + if let Some(generation) = info.start_time { + for entry in &mut info.unresolved_entries { + entry.source_generation = generation; + } + } info.mark_progress_saved(); } @@ -640,7 +661,7 @@ pub(crate) fn pool_meta_has_active_decommission(meta: &PoolMeta) -> bool { } fn is_decommission_suspended(info: &PoolDecommissionInfo) -> bool { - info.has_decommission_state() && !info.queued + info.has_decommission_state() && (!info.queued || !info.unresolved_entries.is_empty()) } fn validate_decommission_terminal_state(complete: bool, failed: bool, canceled: bool) -> Result<()> { @@ -661,6 +682,7 @@ fn invalid_decommission_pool_index_error(pool_count: usize, idx: usize) -> Error enum DecommissionStartPoolState { Missing, Active, + Retryable, Decommissioning, Decommissioned, Blocked, @@ -680,7 +702,11 @@ fn decommission_start_pool_state(pool: Option<&PoolStatus>) -> DecommissionStart if info.complete { DecommissionStartPoolState::Decommissioned } else if info.failed || info.canceled { - DecommissionStartPoolState::Blocked + if info.unresolved_entries.is_empty() { + DecommissionStartPoolState::Blocked + } else { + DecommissionStartPoolState::Retryable + } } else { DecommissionStartPoolState::Decommissioning } @@ -693,7 +719,7 @@ fn is_decommission_start_active_pool(pool: &PoolStatus) -> bool { fn ensure_decommission_start_allowed(state: DecommissionStartPoolState) -> Result<()> { match state { DecommissionStartPoolState::Missing => Err(Error::other("failed to start decommission: target pool was not found")), - DecommissionStartPoolState::Active => Ok(()), + DecommissionStartPoolState::Active | DecommissionStartPoolState::Retryable => Ok(()), DecommissionStartPoolState::Decommissioning => Err(StorageError::DecommissionAlreadyRunning), DecommissionStartPoolState::Decommissioned => { Err(Error::other("failed to start decommission: target pool is already decommissioned")) @@ -710,7 +736,11 @@ fn ensure_decommission_start_keeps_active_pool(meta: &PoolMeta, indices: &[usize .iter() .filter(|pool| is_decommission_start_active_pool(pool)) .count(); - if active_count <= indices.len() { + let active_target_count = indices + .iter() + .filter(|idx| meta.pools.get(**idx).is_some_and(is_decommission_start_active_pool)) + .count(); + if active_count.saturating_sub(active_target_count) == 0 { return Err(Error::other( "failed to start decommission: at least one active pool must remain after decommission start", )); @@ -844,6 +874,107 @@ fn ensure_decommission_generation(meta: &PoolMeta, idx: usize, generation: Offse } } +fn record_decommission_unresolved_entry( + meta: &mut PoolMeta, + idx: usize, + generation: OffsetDateTime, + entry: DecommissionUnresolvedEntry, +) -> Result { + ensure_decommission_generation(meta, idx, generation)?; + if entry.pool_index != idx || entry.source_generation != generation { + return Err(Error::other("decommission unresolved entry does not match the active pool generation")); + } + + let pool_count = meta.pools.len(); + let Some(pool) = meta.pools.get_mut(idx) else { + return Err(invalid_decommission_pool_index_error(pool_count, idx)); + }; + let Some(info) = pool.decommission.as_mut() else { + return Err(decommission_metadata_not_initialized_error("record decommission unresolved entry")); + }; + let existing = info.unresolved_entries.iter_mut().find(|existing| { + existing.bucket == entry.bucket + && existing.object == entry.object + && existing.pool_index == entry.pool_index + && existing.set_index == entry.set_index + && existing.source_generation == entry.source_generation + }); + let changed = match existing { + Some(existing) if existing == &entry => false, + Some(existing) => { + *existing = entry; + true + } + None => { + info.unresolved_entries.push(entry); + true + } + }; + if changed { + pool.last_update = OffsetDateTime::now_utc(); + } + Ok(changed) +} + +type DecommissionUnresolvedEntryIdentity = (usize, String, String); + +fn decommission_unresolved_entry_identity(entry: &DecommissionUnresolvedEntry) -> DecommissionUnresolvedEntryIdentity { + (entry.set_index, entry.bucket.clone(), entry.object.clone()) +} + +fn reconcile_decommission_unresolved_entries_for_completion( + meta: &mut PoolMeta, + idx: usize, + verified_generation: Option, + verified_entries: Option<&[DecommissionUnresolvedEntry]>, +) -> Result<()> { + let pool_count = meta.pools.len(); + let Some(pool) = meta.pools.get(idx) else { + return Err(invalid_decommission_pool_index_error(pool_count, idx)); + }; + let Some(info) = pool.decommission.as_ref() else { + return Err(decommission_metadata_not_initialized_error("reconcile decommission unresolved entries")); + }; + if info.unresolved_entries.is_empty() { + return Ok(()); + } + + let Some(generation) = verified_generation else { + return Err(Error::other(format!( + "failed to complete decommission for pool {idx}: {} unresolved listing entries remain", + info.unresolved_entries.len() + ))); + }; + ensure_decommission_generation(meta, idx, generation)?; + if info + .unresolved_entries + .iter() + .any(|entry| entry.source_generation != generation) + { + return Err(Error::other(format!( + "failed to complete decommission for pool {idx}: unresolved listing ledger contains a different generation" + ))); + } + + let verified_entries = verified_entries.unwrap_or_default(); + let unverified_count = info + .unresolved_entries + .iter() + .filter(|entry| !verified_entries.contains(entry)) + .count(); + if unverified_count > 0 { + return Err(Error::other(format!( + "failed to complete decommission for pool {idx}: {unverified_count} unresolved listing entries were not individually verified" + ))); + } + + let Some(info) = meta.pools.get_mut(idx).and_then(|pool| pool.decommission.as_mut()) else { + return Err(decommission_metadata_not_initialized_error("reconcile decommission unresolved entries")); + }; + info.unresolved_entries.clear(); + Ok(()) +} + async fn run_decommission_side_effect( rx: &CancellationToken, operation_gate: &Arc>, @@ -1023,21 +1154,22 @@ fn resolve_decommission_listing_error(listing_error: Option, entry_error: } } -fn decommission_unresolved_listing_error( - bucket: &str, - prefix: &str, - candidate: Option<&str>, - candidate_count: usize, - disk_error_count: usize, - pool_index: usize, - set_index: usize, -) -> Error { - let location = candidate.unwrap_or(prefix); +fn decommission_unresolved_listing_error(entry: &DecommissionUnresolvedEntry) -> Error { Error::other(format!( - "decommission listing could not resolve metadata for {bucket}/{location} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))" + "decommission listing could not resolve metadata for {bucket}/{object} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))", + bucket = entry.bucket, + object = entry.object, + pool_index = entry.pool_index, + set_index = entry.set_index, + candidate_count = entry.candidate_count, + disk_error_count = entry.disk_error_count, )) } +// The unresolved-entry payload carries listing context by design; boxing it +// would churn every caller without changing behavior. +#[allow(clippy::result_large_err)] +#[allow(clippy::too_many_arguments)] fn resolve_decommission_partial_listing_entry( entries: MetaCacheEntries, resolver: MetadataResolutionParams, @@ -1046,22 +1178,32 @@ fn resolve_decommission_partial_listing_entry( disk_error_count: usize, pool_index: usize, set_index: usize, -) -> Result { + source_generation: OffsetDateTime, +) -> std::result::Result { let candidate_count = entries.as_ref().iter().flatten().count(); if let Some(entry) = entries.resolve(resolver) { return Ok(entry); } - let candidate = entries.as_ref().iter().flatten().map(|entry| entry.name.as_str()).next(); - Err(decommission_unresolved_listing_error( - bucket, - prefix, - candidate, + let object = entries + .as_ref() + .iter() + .flatten() + .map(|entry| entry.name.as_str()) + .next() + .unwrap_or(prefix) + .to_string(); + Err(DecommissionUnresolvedEntry { + bucket: bucket.to_string(), + object, candidate_count, disk_error_count, pool_index, set_index, - )) + source_generation, + observed_at: OffsetDateTime::now_utc(), + reason: "metadata_resolution_failed".to_string(), + }) } fn validate_decommission_durable_ilm_copy( @@ -1484,15 +1626,20 @@ pub(crate) fn pool_meta_movement_snapshot_changed(before: &PoolMeta, after: &Poo /// queued/terminal progressions. Returns whether any entry was replaced or /// appended. pub(crate) fn merge_pool_status_refresh(current: &mut PoolMeta, persisted: PoolMeta, active_workers: &[bool]) -> bool { + let observed_version = current.version.max(persisted.version); if persisted.pools.is_empty() { + current.version = observed_version; return false; } if current.pools.is_empty() { *current = persisted; + current.version = observed_version; return true; } + current.version = observed_version; + let mut merged_newer = false; for (idx, persisted_pool) in persisted.pools.into_iter().enumerate() { if persisted_pool.id != idx { @@ -2402,6 +2549,7 @@ fn ensure_decommission_clear_allowed( complete: bool, failed: bool, canceled: bool, + unresolved_entries: usize, ) -> Result<()> { if !pool_present { return Err(Error::other("failed to clear decommission: target pool was not found")); @@ -2419,6 +2567,12 @@ fn ensure_decommission_clear_allowed( return Err(StorageError::DecommissionAlreadyRunning); } + if unresolved_entries > 0 { + return Err(Error::other(format!( + "failed to clear decommission: {unresolved_entries} unresolved listing entries must be reconciled by retrying decommission" + ))); + } + Ok(()) } @@ -2533,36 +2687,43 @@ fn decode_pool_meta_replica(data: Vec) -> PoolMetaReplica { return PoolMetaReplica::Incompatible(format!("unsupported format {format}")); } let version = LittleEndian::read_u16(&data[2..4]); - if version != POOL_META_VERSION { + if !matches!(version, POOL_META_V1_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) { + let meta = match (version, rmp::decode::read_array_len(&mut &payload[..])) { + (POOL_META_VERSION, 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) { + (POOL_META_V1_VERSION, 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("legacy", err), + Err(err) => return classify_pool_meta_tuple_decode_error("v1", err), }, - Ok(field_count) if field_count < 2 => { + // The older V1 tuple includes the runtime-only `dont_save` flag. + (POOL_META_V1_VERSION, 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 v1", 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}")); + (_, Ok(field_count)) => { + return PoolMetaReplica::Incompatible(format!( + "pool metadata version {version} tuple has unsupported field count {field_count}" + )); } - Err(_) => { + (_, Err(_)) => { let mut meta = PoolMeta::default(); if let Err(err) = meta.load_from_config_data(data.clone()) { let reason = err.to_string(); @@ -2575,7 +2736,7 @@ fn decode_pool_meta_replica(data: Vec) -> PoolMetaReplica { } }; - match meta.encode_config_data() { + match meta.encode_config_data_for_v2_gate(true) { Ok(canonical) => PoolMetaReplica::Valid { raw: data, canonical, @@ -2608,15 +2769,16 @@ fn select_pool_meta_replica(replicas: Vec) -> Result, Vec, PoolMeta)> = None; let mut needs_repair = false; let mut repair_write_safe = true; let mut missing = 0usize; let mut unusable = Vec::new(); + let mut observed_version = POOL_META_V1_VERSION; for (idx, replica) in replicas.into_iter().enumerate() { match replica { @@ -2639,6 +2801,7 @@ fn select_pool_meta_replica(replicas: Vec) -> Result { + observed_version = observed_version.max(meta.version); if let Some((selected_idx, selected_raw, selected_canonical, _)) = selected.as_ref() { if selected_canonical != &canonical { if *selected_idx == 0 { @@ -2661,7 +2824,8 @@ fn select_pool_meta_replica(replicas: Vec) -> Result, #[serde(rename = "terminalReloadFailures", default)] pub terminal_reload_failures: Vec, + #[serde(rename = "unresolvedEntries", default)] + pub unresolved_entries: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedPoolMetaV1 { + pub version: u16, + pub pools: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedPoolStatusV1 { + #[serde(rename = "id")] + pub id: usize, + #[serde(rename = "cmdline")] + pub cmd_line: String, + #[serde(rename = "lastUpdate", with = "time::serde::rfc3339")] + pub last_update: OffsetDateTime, + #[serde(rename = "decommissionInfo")] + pub decommission: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedPoolDecommissionInfoV1 { + #[serde(rename = "startTime", with = "time::serde::rfc3339::option")] + pub start_time: Option, + #[serde(rename = "startSize")] + pub start_size: usize, + #[serde(rename = "totalSize")] + pub total_size: usize, + #[serde(rename = "currentSize")] + pub current_size: usize, + #[serde(rename = "complete")] + pub complete: bool, + #[serde(rename = "failed")] + pub failed: bool, + #[serde(rename = "canceled")] + pub canceled: bool, + #[serde(rename = "queued", default)] + pub queued: bool, + #[serde(rename = "queuedBuckets", default)] + pub queued_buckets: Vec, + #[serde(rename = "decommissionedBuckets", default)] + pub decommissioned_buckets: Vec, + #[serde(rename = "bucket", default)] + pub bucket: String, + #[serde(rename = "prefix", default)] + pub prefix: String, + #[serde(rename = "object", default)] + pub object: String, + #[serde(rename = "objectsDecommissioned")] + pub items_decommissioned: usize, + #[serde(rename = "objectsDecommissionedFailed")] + pub items_decommission_failed: usize, + #[serde(rename = "bytesDecommissioned")] + pub bytes_done: usize, + #[serde(rename = "bytesDecommissionedFailed")] + pub bytes_failed: usize, + #[serde(rename = "terminalReloadAttemptAt", with = "time::serde::rfc3339::option", default)] + pub terminal_reload_attempt_at: Option, + #[serde(rename = "terminalReloadFailures", default)] + pub terminal_reload_failures: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -2805,12 +3034,35 @@ struct LegacyPoolDecommissionInfo { pub bytes_failed: usize, } +fn ensure_pool_meta_payload_version(actual: u16, expected: u16, kind: &str) -> Result<()> { + if actual == expected { + return Ok(()); + } + Err(Error::other(format!( + "pool metadata {kind} payload has version {actual}, expected {expected}" + ))) +} + impl TryFrom for PoolMeta { type Error = Error; fn try_from(value: PersistedPoolMeta) -> Result { + ensure_pool_meta_payload_version(value.version, POOL_META_VERSION, "current")?; Ok(Self { - version: value.version, + version: POOL_META_VERSION, + pools: value.pools.into_iter().map(TryInto::try_into).collect::>>()?, + dont_save: false, + }) + } +} + +impl TryFrom for PoolMeta { + type Error = Error; + + fn try_from(value: PersistedPoolMetaV1) -> Result { + ensure_pool_meta_payload_version(value.version, POOL_META_V1_VERSION, "v1")?; + Ok(Self { + version: POOL_META_V1_VERSION, pools: value.pools.into_iter().map(TryInto::try_into).collect::>>()?, dont_save: false, }) @@ -2826,8 +3078,9 @@ impl TryFrom for PoolMeta { pools, dont_save: _, } = value; + ensure_pool_meta_payload_version(version, POOL_META_V1_VERSION, "legacy v1")?; Ok(Self { - version, + version: POOL_META_V1_VERSION, pools: pools.into_iter().map(TryInto::try_into).collect::>>()?, dont_save: false, }) @@ -2847,6 +3100,19 @@ impl TryFrom for PoolStatus { } } +impl TryFrom for PoolStatus { + type Error = Error; + + fn try_from(value: PersistedPoolStatusV1) -> Result { + Ok(Self { + id: value.id, + cmd_line: value.cmd_line, + last_update: value.last_update, + decommission: value.decommission.map(TryInto::try_into).transpose()?, + }) + } +} + impl TryFrom for PoolStatus { type Error = Error; @@ -2886,6 +3152,40 @@ impl TryFrom for PoolDecommissionInfo { bytes_failed: value.bytes_failed, terminal_reload_attempt_at: value.terminal_reload_attempt_at, terminal_reload_failures: value.terminal_reload_failures, + unresolved_entries: value.unresolved_entries, + progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed), + progress_save_retry_after: None, + }) + } +} + +impl TryFrom for PoolDecommissionInfo { + type Error = Error; + + fn try_from(value: PersistedPoolDecommissionInfoV1) -> Result { + validate_decommission_terminal_state(value.complete, value.failed, value.canceled)?; + Ok(Self { + start_time: value.start_time, + start_size: value.start_size, + total_size: value.total_size, + current_size: value.current_size, + complete: value.complete, + failed: value.failed, + canceled: value.canceled, + queued: value.queued, + queued_buckets: value.queued_buckets, + decommissioned_buckets: value.decommissioned_buckets, + bucket: value.bucket, + prefix: value.prefix, + object: value.object, + stage: String::new(), + items_decommissioned: value.items_decommissioned, + items_decommission_failed: value.items_decommission_failed, + bytes_done: value.bytes_done, + bytes_failed: value.bytes_failed, + terminal_reload_attempt_at: value.terminal_reload_attempt_at, + terminal_reload_failures: value.terminal_reload_failures, + unresolved_entries: Vec::new(), progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed), progress_save_retry_after: None, }) @@ -2918,6 +3218,7 @@ impl TryFrom for PoolDecommissionInfo { bytes_failed: value.bytes_failed, terminal_reload_attempt_at: None, terminal_reload_failures: Vec::new(), + unresolved_entries: Vec::new(), progress_save_item_baseline: value.items_decommissioned.saturating_add(value.items_decommission_failed), progress_save_retry_after: None, }) @@ -2927,7 +3228,16 @@ impl TryFrom for PoolDecommissionInfo { impl From<&PoolMeta> for PersistedPoolMeta { fn from(value: &PoolMeta) -> Self { Self { - version: value.version, + version: POOL_META_VERSION, + pools: value.pools.iter().map(Into::into).collect(), + } + } +} + +impl From<&PoolMeta> for PersistedPoolMetaV1 { + fn from(value: &PoolMeta) -> Self { + Self { + version: POOL_META_V1_VERSION, pools: value.pools.iter().map(Into::into).collect(), } } @@ -2944,7 +3254,45 @@ impl From<&PoolStatus> for PersistedPoolStatus { } } +impl From<&PoolStatus> for PersistedPoolStatusV1 { + fn from(value: &PoolStatus) -> Self { + Self { + id: value.id, + cmd_line: value.cmd_line.clone(), + last_update: value.last_update, + decommission: value.decommission.as_ref().map(Into::into), + } + } +} + impl From<&PoolDecommissionInfo> for PersistedPoolDecommissionInfo { + fn from(value: &PoolDecommissionInfo) -> Self { + Self { + start_time: value.start_time, + start_size: value.start_size, + total_size: value.total_size, + current_size: value.current_size, + complete: value.complete, + failed: value.failed, + canceled: value.canceled, + queued: value.queued, + queued_buckets: value.queued_buckets.clone(), + decommissioned_buckets: value.decommissioned_buckets.clone(), + bucket: value.bucket.clone(), + prefix: value.prefix.clone(), + object: value.object.clone(), + items_decommissioned: value.items_decommissioned, + items_decommission_failed: value.items_decommission_failed, + bytes_done: value.bytes_done, + bytes_failed: value.bytes_failed, + terminal_reload_attempt_at: value.terminal_reload_attempt_at, + terminal_reload_failures: value.terminal_reload_failures.clone(), + unresolved_entries: value.unresolved_entries.clone(), + } + } +} + +impl From<&PoolDecommissionInfo> for PersistedPoolDecommissionInfoV1 { fn from(value: &PoolDecommissionInfo) -> Self { Self { start_time: value.start_time, @@ -2971,23 +3319,33 @@ impl From<&PoolDecommissionInfo> for PersistedPoolDecommissionInfo { } impl PoolMeta { - fn decode_pool_meta_payload(payload: &[u8]) -> Result { - match rmp_serde::from_slice::(payload) { - Ok(meta) => meta.try_into(), - Err(persisted_err) => { - let legacy: LegacyPoolMeta = rmp_serde::from_slice(payload).map_err(|legacy_err| { - Error::other(format!( - "PoolMeta decode failed for both persisted and legacy formats: persisted={persisted_err}; legacy={legacy_err}" - )) - })?; - legacy.try_into() - } + fn decode_pool_meta_payload(version: u16, payload: &[u8]) -> Result { + match version { + POOL_META_VERSION => rmp_serde::from_slice::(payload) + .map_err(|err| Error::other(format!("PoolMeta v{POOL_META_VERSION} decode failed: {err}")))? + .try_into(), + POOL_META_V1_VERSION => match rmp_serde::from_slice::(payload) { + Ok(meta) => meta.try_into(), + Err(persisted_err) => { + let legacy: LegacyPoolMeta = rmp_serde::from_slice(payload).map_err(|legacy_err| { + Error::other(format!( + "PoolMeta v1 decode failed for both persisted and legacy formats: persisted={persisted_err}; legacy={legacy_err}" + )) + })?; + legacy.try_into() + } + }, + _ => Err(Error::other(format!("pool metadata load failed: unknown version {version}"))), } } pub fn new(pools: &[Arc], prev_meta: &PoolMeta) -> Self { let mut new_meta = Self { - version: POOL_META_VERSION, + version: if prev_meta.version == POOL_META_VERSION || pool_meta_v2_writer_enabled() { + POOL_META_VERSION + } else { + POOL_META_V1_VERSION + }, pools: Vec::new(), ..Default::default() }; @@ -3121,13 +3479,13 @@ impl PoolMeta { return Err(Error::other(format!("pool metadata load failed: unknown format {format}"))); } let version = LittleEndian::read_u16(&data[2..4]); - if version != POOL_META_VERSION { + if !matches!(version, POOL_META_V1_VERSION | POOL_META_VERSION) { return Err(Error::other(format!("pool metadata load failed: unknown version {version}"))); } - *self = Self::decode_pool_meta_payload(&data[4..])?; + *self = Self::decode_pool_meta_payload(version, &data[4..])?; - if self.version != POOL_META_VERSION { + if !matches!(self.version, POOL_META_V1_VERSION | POOL_META_VERSION) { return Err(Error::other(format!( "pool metadata load failed: unexpected decoded version {}", self.version @@ -3154,18 +3512,55 @@ impl PoolMeta { } fn encode_config_data(&self) -> Result> { + self.encode_config_data_for_v2_gate(pool_meta_v2_writer_enabled()) + } + + fn encode_config_data_for_v2_gate(&self, v2_enabled: bool) -> Result> { if self.dont_save { return Ok(Vec::new()); } + if !matches!(self.version, 0 | POOL_META_V1_VERSION | POOL_META_VERSION) { + return Err(Error::other(format!( + "pool metadata save failed: unexpected runtime version {}", + self.version + ))); + } + let version = if self.version == POOL_META_VERSION || v2_enabled { + POOL_META_VERSION + } else { + POOL_META_V1_VERSION + }; + if version == POOL_META_V1_VERSION + && self + .pools + .iter() + .filter_map(|pool| pool.decommission.as_ref()) + .any(|info| !info.unresolved_entries.is_empty()) + { + return Err(Error::other(format!( + "pool metadata V2 is required to persist unresolved decommission entries; enable both {} and {} only after every reader and writer supports V2", + rustfs_config::ENV_POOL_META_V2_WRITE, + rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED, + ))); + } let mut data = Vec::new(); data.write_u16::(POOL_META_FORMAT)?; - data.write_u16::(POOL_META_VERSION)?; + data.write_u16::(version)?; let mut buf = Vec::new(); - PersistedPoolMeta::from(self).serialize(&mut Serializer::new(&mut buf))?; + match version { + POOL_META_V1_VERSION => PersistedPoolMetaV1::from(self).serialize(&mut Serializer::new(&mut buf))?, + POOL_META_VERSION => PersistedPoolMeta::from(self).serialize(&mut Serializer::new(&mut buf))?, + _ => unreachable!("pool metadata writer selected an unsupported version"), + } data.write_all(&buf)?; Ok(data) } + #[cfg(test)] + pub(crate) fn encode_config_data_for_test(&self) -> Result> { + self.encode_config_data_for_v2_gate(true) + } + pub async fn save(&self, pools: Vec>) -> Result<()> { let data = self.encode_config_data()?; if data.is_empty() { @@ -3333,13 +3728,21 @@ impl PoolMeta { return Err(invalid_decommission_pool_index_error(pool_count, idx)); }; - let (decommission_present, complete, failed, canceled) = pool + let (decommission_present, complete, failed, canceled, unresolved_entries) = pool .decommission .as_ref() - .map(|info| (info.has_decommission_state(), info.complete, info.failed, info.canceled)) - .unwrap_or((false, false, false, false)); + .map(|info| { + ( + info.has_decommission_state(), + info.complete, + info.failed, + info.canceled, + info.unresolved_entries.len(), + ) + }) + .unwrap_or((false, false, false, false, 0)); - ensure_decommission_clear_allowed(true, decommission_present, complete, failed, canceled)?; + ensure_decommission_clear_allowed(true, decommission_present, complete, failed, canceled, unresolved_entries)?; pool.last_update = OffsetDateTime::now_utc(); pool.decommission = None; @@ -3428,7 +3831,10 @@ impl PoolMeta { let now = OffsetDateTime::now_utc(); pool.last_update = now; info.queued = false; - info.start_time.get_or_insert(now); + let generation = *info.start_time.get_or_insert(now); + for entry in &mut info.unresolved_entries { + entry.source_generation = generation; + } return true; } @@ -3593,6 +3999,26 @@ pub fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (Strin path_to_bucket_object_with_base_path(base_path, path) } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DecommissionUnresolvedEntry { + pub bucket: String, + pub object: String, + #[serde(rename = "poolIndex")] + pub pool_index: usize, + #[serde(rename = "setIndex")] + pub set_index: usize, + #[serde(rename = "sourceGeneration", with = "time::serde::rfc3339")] + pub source_generation: OffsetDateTime, + #[serde(rename = "candidateCount")] + pub candidate_count: usize, + #[serde(rename = "diskErrorCount")] + pub disk_error_count: usize, + #[serde(rename = "observedAt", with = "time::serde::rfc3339")] + pub observed_at: OffsetDateTime, + pub reason: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct PoolDecommissionInfo { #[serde(rename = "startTime", with = "time::serde::rfc3339::option")] @@ -3638,6 +4064,8 @@ pub struct PoolDecommissionInfo { #[serde(rename = "terminalReloadFailures", default)] pub terminal_reload_failures: Vec, #[serde(skip)] + pub unresolved_entries: Vec, + #[serde(skip)] pub progress_save_item_baseline: usize, #[serde(skip)] pub progress_save_retry_after: Option, @@ -3671,6 +4099,7 @@ impl PoolDecommissionInfo { || self.bytes_failed > 0 || self.terminal_reload_attempt_at.is_some() || !self.terminal_reload_failures.is_empty() + || !self.unresolved_entries.is_empty() } fn counted_items(&self) -> usize { @@ -4112,6 +4541,21 @@ impl ECStore { snapshot.save(self.pools.clone()).await } + async fn persist_decommission_unresolved_entry( + &self, + idx: usize, + generation: OffsetDateTime, + entry: DecommissionUnresolvedEntry, + ) -> Result<()> { + { + let mut pool_meta = self.pool_meta.write().await; + record_decommission_unresolved_entry(&mut pool_meta, idx, generation, entry)?; + } + self.save_current_pool_meta() + .await + .map_err(|err| Error::other(format!("decommission unresolved entry ledger save failed: {err}"))) + } + async fn save_decommission_progress_checkpoint(&self, idx: usize, generation: OffsetDateTime) -> Result { // Lock order: save gate, then the short pool metadata read/write sections. Peer // reloads are intentionally performed by the caller after both locks are released. @@ -4565,12 +5009,20 @@ impl ECStore { let Some(pool) = pool_meta.pools.get(idx) else { return Err(invalid_decommission_pool_index_error(pool_count, idx)); }; - let (decommission_present, complete, failed, canceled) = pool + let (decommission_present, complete, failed, canceled, unresolved_entries) = pool .decommission .as_ref() - .map(|info| (info.has_decommission_state(), info.complete, info.failed, info.canceled)) - .unwrap_or((false, false, false, false)); - ensure_decommission_clear_allowed(true, decommission_present, complete, failed, canceled)?; + .map(|info| { + ( + info.has_decommission_state(), + info.complete, + info.failed, + info.canceled, + info.unresolved_entries.len(), + ) + }) + .unwrap_or((false, false, false, false, 0)); + ensure_decommission_clear_allowed(true, decommission_present, complete, failed, canceled, unresolved_entries)?; } // Cancel workers before waiting for the movement writer so active // object operations can observe the signal and release read guards. @@ -5111,6 +5563,7 @@ impl ECStore { let list_bi = bi.clone(); let list_outstanding = outstanding.clone(); let list_entry_error = entry_error.clone(); + let list_store = self.clone(); let mut listing = tokio::spawn(async move { run_decommission_listing_with_retry_and_drain( list_rx.clone(), @@ -5124,9 +5577,20 @@ impl ECStore { let rx = list_rx_for_list.clone(); let bucket = list_bi.clone(); let entry_error = list_entry_error.clone(); + let store = list_store.clone(); async move { - set.list_objects_to_decommission(rx, bucket, callback, entry_error, idx, set_idx) - .await + set.list_objects_to_decommission( + store, + rx, + bucket, + callback, + entry_error, + idx, + set_idx, + generation, + false, + ) + .await } }, move || { @@ -6562,19 +7026,22 @@ impl ECStore { state = "verifying_completion", "Decommission completion verification started" ); - if let Err(err) = self.check_after_decommission(idx, &rx, generation).await { - if is_err_operation_canceled(&err) { - return Err(err); + let verified_unresolved_entries = match self.check_after_decommission(idx, &rx, generation).await { + Ok(verified_entries) => verified_entries, + Err(err) => { + if is_err_operation_canceled(&err) { + return Err(err); + } + resolve_decommission_terminal_mark_result( + self.decommission_failed_for_operation(idx, canceler).await, + "failed", + &cmd_line, + )?; + return Err(Error::other(format!( + "failed to finalize decommission for pool {cmd_line}: post-check failed: {err}" + ))); } - resolve_decommission_terminal_mark_result( - self.decommission_failed_for_operation(idx, canceler).await, - "failed", - &cmd_line, - )?; - return Err(Error::other(format!( - "failed to finalize decommission for pool {cmd_line}: post-check failed: {err}" - ))); - } + }; info!( event = EVENT_DECOMMISSION_STATE, component = LOG_COMPONENT_ECSTORE, @@ -6584,7 +7051,10 @@ impl ECStore { state = "marking_completed", "Decommission marking completed state" ); - if let Err(err) = self.complete_decommission_for_operation(idx, canceler).await { + if let Err(err) = self + .complete_decommission_for_operation(idx, canceler, generation, verified_unresolved_entries) + .await + { resolve_decommission_terminal_mark_result( self.decommission_failed_for_operation(idx, canceler).await, "failed", @@ -6733,14 +7203,27 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn complete_decommission(&self, idx: usize) -> Result<()> { - self.complete_decommission_with_owner(idx, None).await + self.complete_decommission_with_owner(idx, None, None, None).await } - async fn complete_decommission_for_operation(&self, idx: usize, owner: &DecommissionCanceler) -> Result<()> { - self.complete_decommission_with_owner(idx, Some(owner)).await + async fn complete_decommission_for_operation( + &self, + idx: usize, + owner: &DecommissionCanceler, + verified_generation: OffsetDateTime, + verified_unresolved_entries: Vec, + ) -> Result<()> { + self.complete_decommission_with_owner(idx, Some(owner), Some(verified_generation), Some(verified_unresolved_entries)) + .await } - async fn complete_decommission_with_owner(&self, idx: usize, owner: Option<&DecommissionCanceler>) -> Result<()> { + async fn complete_decommission_with_owner( + &self, + idx: usize, + owner: Option<&DecommissionCanceler>, + verified_generation: Option, + verified_unresolved_entries: Option>, + ) -> Result<()> { ensure_decommission_terminal_operation_supported(self.single_pool(), "complete decommission")?; ensure_valid_decommission_pool_index(self.pools.len(), idx)?; if let Some(owner) = owner { @@ -6764,11 +7247,18 @@ impl ECStore { let previous_pool_meta = pool_meta.clone(); let Some(changed) = update_decommission_for_operation(cancelers.as_slice(), &mut pool_meta, idx, owner, |pool_meta| { - pool_meta.decommission_complete(idx) + reconcile_decommission_unresolved_entries_for_completion( + pool_meta, + idx, + verified_generation, + verified_unresolved_entries.as_deref(), + )?; + Ok::(pool_meta.decommission_complete(idx)) }) else { return Ok(()); }; + let changed = changed?; let completed = pool_meta .pools .get(idx) @@ -8075,13 +8565,42 @@ impl ECStore { idx: usize, rx: &CancellationToken, generation: OffsetDateTime, - ) -> Result<()> { + ) -> Result> { self.ensure_decommission_generation_current(idx, generation).await?; let operation_gate = self.ctx.data_movement_operation_gate(); - run_decommission_side_effect(rx, &operation_gate, || self.check_after_decommission_unfenced(idx)).await + run_decommission_side_effect(rx, &operation_gate, || self.check_after_decommission_unfenced(idx, generation)).await } - async fn check_after_decommission_unfenced(self: &Arc, idx: usize) -> Result<()> { + async fn check_after_decommission_unfenced( + self: &Arc, + idx: usize, + generation: OffsetDateTime, + ) -> Result> { + let unresolved_entries = { + let pool_meta = self.pool_meta.read().await; + ensure_decommission_generation(&pool_meta, idx, generation)?; + let info = pool_meta.pools[idx] + .decommission + .as_ref() + .ok_or_else(|| decommission_metadata_not_initialized_error("verify unresolved decommission entries"))?; + if info + .unresolved_entries + .iter() + .any(|entry| entry.pool_index != idx || entry.source_generation != generation) + { + return Err(Error::other(format!( + "failed to verify decommission for pool {idx}: unresolved listing ledger contains a different pool or generation" + ))); + } + info.unresolved_entries.clone() + }; + let unresolved_entries_by_identity = Arc::new( + unresolved_entries + .iter() + .map(|entry| (decommission_unresolved_entry_identity(entry), entry.clone())) + .collect::>(), + ); + let resolved_unresolved_entries = Arc::new(tokio::sync::Mutex::new(Vec::new())); let buckets = self.get_buckets_to_decommission().await?; let pool = self.pools[idx].clone(); @@ -8089,13 +8608,21 @@ impl ECStore { .await?; for (set_index, set) in pool.disk_set.iter().enumerate() { + let require_all_disks = unresolved_entries.iter().any(|entry| entry.set_index == set_index); for bucket_info in &buckets { let mut lifecycle_config = None; let mut object_lock_config = None; + let mut replication_configured = false; if bucket_info.name != RUSTFS_META_BUCKET { let expiry_configs = get_expiry_configs(self, &bucket_info.name).await?; lifecycle_config = expiry_configs.lifecycle.map(|config| (*config).clone()); object_lock_config = expiry_configs.object_lock.map(|config| (*config).clone()); + replication_configured = resolve_decommission_optional_bucket_config_result( + &bucket_info.name, + "replication", + metadata_sys::get_replication_config(&bucket_info.name).await, + )? + .is_some(); } let versions_found = Arc::new(AtomicUsize::new(0)); @@ -8108,9 +8635,12 @@ impl ECStore { let bucket_name = bucket_info.name.clone(); let lifecycle_config_cb = lifecycle_config.clone(); let object_lock_config_cb = object_lock_config.clone(); + let replication_configured_cb = replication_configured; let store = Arc::clone(self); let source_set = set.clone(); let callback_rx_cb = callback_rx.clone(); + let unresolved_entries_by_identity_cb = unresolved_entries_by_identity.clone(); + let resolved_unresolved_entries_cb = resolved_unresolved_entries.clone(); let callback: ListCallback = Arc::new(move |entry: MetaCacheEntry| { let versions_found = versions_found_cb.clone(); @@ -8119,9 +8649,12 @@ impl ECStore { let bucket_name = bucket_name.clone(); let lifecycle_config = lifecycle_config_cb.clone(); let object_lock_config = object_lock_config_cb.clone(); + let replication_configured = replication_configured_cb; let store = Arc::clone(&store); let source_set = source_set.clone(); let callback_rx = callback_rx_cb.clone(); + let unresolved_entries_by_identity = unresolved_entries_by_identity_cb.clone(); + let resolved_unresolved_entries = resolved_unresolved_entries_cb.clone(); Box::pin(async move { if callback_rx.is_cancelled() { return; @@ -8170,7 +8703,7 @@ impl ECStore { return; } - let fivs = match load_decommission_entry_exact_versions( + let mut fivs = match load_decommission_entry_exact_versions( &source_set, &entry, &bucket_name, @@ -8189,7 +8722,11 @@ impl ECStore { } }; + fivs.versions + .sort_by_key(|version| (version.mod_time.is_none(), std::cmp::Reverse(version.mod_time))); + let mut remaining = 0; + let mut expired = 0; for version in fivs.versions.iter().chain(fivs.free_versions.iter()) { if version.tier_free_version() { remaining += 1; @@ -8207,9 +8744,6 @@ impl ECStore { ); continue; } - if version.deleted { - continue; - } let skip_lifecycle = match should_skip_lifecycle_for_data_movement( Arc::clone(&store), &bucket_name, @@ -8233,6 +8767,11 @@ impl ECStore { } }; if skip_lifecycle { + expired += 1; + continue; + } + let remaining_versions = decommission_remaining_version_count(&fivs.versions, expired); + if should_skip_decommission_delete_marker(version, remaining_versions, replication_configured) { continue; } remaining += 1; @@ -8243,6 +8782,14 @@ impl ECStore { if first_path.is_none() { *first_path = Some(format!("{bucket_name}/{}", entry.name)); } + } else { + let identity = (set_index, bucket_name.clone(), entry.name.clone()); + if let Some(unresolved_entry) = unresolved_entries_by_identity.get(&identity) { + let mut resolved = resolved_unresolved_entries.lock().await; + if !resolved.contains(unresolved_entry) { + resolved.push(unresolved_entry.clone()); + } + } } versions_found.fetch_add(remaining, Ordering::Relaxed); @@ -8250,7 +8797,17 @@ impl ECStore { }); let list_result = set - .list_objects_to_decommission(callback_rx, bucket_info.clone(), callback, entry_error.clone(), idx, set_index) + .list_objects_to_decommission( + self.clone(), + callback_rx, + bucket_info.clone(), + callback, + entry_error.clone(), + idx, + set_index, + generation, + require_all_disks, + ) .await; let entry_error = entry_error.lock().await.clone(); resolve_decommission_check_after_list_result(list_result, entry_error)?; @@ -8270,10 +8827,31 @@ impl ECStore { } } + let mut verified_unresolved_entries = resolved_unresolved_entries.lock().await.clone(); + for entry in &unresolved_entries { + let set = pool.disk_set.get(entry.set_index).ok_or_else(|| { + Error::other(format!( + "failed to verify decommission for pool {idx}: unresolved listing entry references missing set {}", + entry.set_index + )) + })?; + if set.decommission_unresolved_entry_absent_on_all_disks(idx, entry).await? + && !verified_unresolved_entries.contains(entry) + { + verified_unresolved_entries.push(entry.clone()); + } + if !verified_unresolved_entries.contains(entry) { + return Err(Error::other(format!( + "failed to verify decommission for pool {idx}: unresolved listing entry {}/{} in set {} was neither re-observed as resolved nor absent on every source disk", + entry.bucket, entry.object, entry.set_index + ))); + } + } + self.persist_decommission_durable_ilm_manifest(idx).await?; self.verify_decommission_durable_ilm_receipts(idx).await?; - Ok(()) + Ok(verified_unresolved_entries) } async fn ensure_decommission_multipart_uploads_drained( @@ -8323,6 +8901,7 @@ impl ECStore { let generation = self.active_decommission_generation(idx).await?; self.check_after_decommission(idx, &CancellationToken::new(), generation) .await + .map(|_| ()) } #[tracing::instrument(skip(self, rd))] @@ -8513,18 +9092,43 @@ mod tests { .expect("pool metadata should encode") } - fn pool_meta_legacy_replica_test_data(cmd_line: &str) -> Vec { + fn pool_meta_v1_replica_test_data(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) + data.write_u16::(POOL_META_V1_VERSION) .expect("pool metadata version should encode"); - pool_meta_replica_test_meta(cmd_line) + PersistedPoolMetaV1::from(meta) .serialize(&mut Serializer::new(&mut data)) .expect("legacy pool metadata should encode"); data } + fn pool_meta_persisted_v1_replica_test_data(cmd_line: &str) -> Vec { + pool_meta_v1_replica_test_data(&pool_meta_replica_test_meta(cmd_line)) + } + + fn pool_meta_legacy_v1_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_V1_VERSION) + .expect("pool metadata version should encode"); + LegacyPoolMeta { + version: POOL_META_V1_VERSION, + pools: vec![LegacyPoolStatus { + id: 0, + cmd_line: cmd_line.to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }], + dont_save: false, + } + .serialize(&mut Serializer::new(&mut data)) + .expect("legacy v1 pool metadata should encode"); + data + } + #[test] fn pool_meta_replica_selection_falls_back_from_corrupt_first_copy() { let selection = select_pool_meta_replica(vec![ @@ -8580,7 +9184,7 @@ mod tests { #[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_legacy_v1_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"); @@ -8590,6 +9194,156 @@ mod tests { assert_eq!(selection.meta.pools[0].cmd_line, "pool-0"); } + #[test] + fn pool_meta_v1_replica_migrates_to_v2_canonical_form() { + let mut source = pool_meta_replica_test_meta("pool-0"); + source.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + queued_buckets: vec!["bucket-a".to_string()], + ..Default::default() + }); + let data = pool_meta_v1_replica_test_data(&source); + let PoolMetaReplica::Valid { raw, canonical, meta } = decode_pool_meta_replica(data) else { + panic!("v1 pool metadata should remain readable"); + }; + + assert_eq!(LittleEndian::read_u16(&raw[2..4]), POOL_META_V1_VERSION); + assert_eq!(LittleEndian::read_u16(&canonical[2..4]), POOL_META_VERSION); + assert_eq!(meta.version, POOL_META_V1_VERSION); + let info = meta.pools[0] + .decommission + .as_ref() + .expect("v1 decommission state should migrate"); + assert_eq!(info.queued_buckets, vec!["bucket-a".to_string()]); + assert!(info.unresolved_entries.is_empty()); + } + + #[test] + fn pool_meta_legacy_v1_replica_migrates_to_v2_canonical_form() { + let data = pool_meta_legacy_v1_replica_test_data("pool-0"); + let PoolMetaReplica::Valid { raw, canonical, meta } = decode_pool_meta_replica(data) else { + panic!("legacy v1 pool metadata should remain readable"); + }; + + assert_eq!(LittleEndian::read_u16(&raw[2..4]), POOL_META_V1_VERSION); + assert_eq!(LittleEndian::read_u16(&canonical[2..4]), POOL_META_VERSION); + assert_eq!(meta.version, POOL_META_V1_VERSION); + assert_eq!(meta.pools[0].cmd_line, "pool-0"); + } + + #[test] + fn pool_meta_replica_selection_rejects_v1_v2_divergence() { + let err = select_pool_meta_replica(vec![ + decode_pool_meta_replica(pool_meta_persisted_v1_replica_test_data("pool-old")), + decode_pool_meta_replica(pool_meta_replica_test_data("pool-new")), + ]) + .expect_err("different v1 and v2 snapshots must remain fail-closed"); + + assert!(err.to_string().contains("valid replicas in pools 0 and 1 diverge")); + } + + #[test] + fn pool_meta_replica_rejects_header_payload_version_mismatch() { + let mut v1_with_v2_header = pool_meta_persisted_v1_replica_test_data("pool-0"); + LittleEndian::write_u16(&mut v1_with_v2_header[2..4], POOL_META_VERSION); + assert!(matches!(decode_pool_meta_replica(v1_with_v2_header), PoolMetaReplica::Corrupt(_))); + + let mut v2_with_v1_header = pool_meta_replica_test_data("pool-0"); + LittleEndian::write_u16(&mut v2_with_v1_header[2..4], POOL_META_V1_VERSION); + assert!(matches!( + decode_pool_meta_replica(v2_with_v1_header), + PoolMetaReplica::Corrupt(_) | PoolMetaReplica::Incompatible(_) + )); + } + + #[test] + fn pool_meta_v2_header_guards_v1_readers_from_tuple_extension() { + let mut meta = pool_meta_replica_test_meta("pool-0"); + meta.pools[0].decommission = Some(PoolDecommissionInfo::default()); + let data = meta.encode_config_data().expect("v2 pool metadata should encode"); + + assert_eq!(LittleEndian::read_u16(&data[2..4]), POOL_META_VERSION); + assert_ne!(POOL_META_VERSION, POOL_META_V1_VERSION); + assert!(rmp_serde::from_slice::(&data[4..]).is_err()); + } + + #[test] + fn pool_meta_writer_stays_v1_until_v2_is_fleet_confirmed() { + let mut meta = pool_meta_replica_test_meta("pool-0"); + meta.version = POOL_META_V1_VERSION; + + let v1 = meta + .encode_config_data_for_v2_gate(false) + .expect("the default writer should preserve v1"); + let v2 = meta + .encode_config_data_for_v2_gate(true) + .expect("the confirmed writer should emit v2"); + + assert_eq!(LittleEndian::read_u16(&v1[2..4]), POOL_META_V1_VERSION); + assert_eq!(LittleEndian::read_u16(&v2[2..4]), POOL_META_VERSION); + assert!(rmp_serde::from_slice::(&v1[4..]).is_ok()); + } + + #[test] + fn pool_meta_v2_writer_requires_both_gates() { + assert!(!pool_meta_v2_writer_enabled_for(false, false)); + assert!(!pool_meta_v2_writer_enabled_for(true, false)); + assert!(!pool_meta_v2_writer_enabled_for(false, true)); + assert!(pool_meta_v2_writer_enabled_for(true, true)); + } + + #[test] + fn pool_meta_v2_floor_is_sticky_after_observation() { + let meta = pool_meta_replica_test_meta("pool-0"); + let data = meta + .encode_config_data_for_v2_gate(false) + .expect("an observed v2 snapshot must not be downgraded"); + + assert_eq!(LittleEndian::read_u16(&data[2..4]), POOL_META_VERSION); + } + + #[test] + fn pool_meta_v1_writer_rejects_unresolved_ledger() { + let mut meta = pool_meta_replica_test_meta("pool-0"); + meta.version = POOL_META_V1_VERSION; + meta.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + unresolved_entries: vec![DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: OffsetDateTime::UNIX_EPOCH, + candidate_count: 1, + disk_error_count: 1, + observed_at: OffsetDateTime::UNIX_EPOCH, + reason: "metadata_resolution_failed".to_string(), + }], + ..Default::default() + }); + + let err = meta + .encode_config_data_for_v2_gate(false) + .expect_err("v1 must not drop the unresolved ledger"); + assert!(err.to_string().contains("Pool metadata V2 is required")); + } + + #[test] + fn pool_meta_replica_selection_preserves_observed_v2_floor() { + let selection = select_pool_meta_replica(vec![ + decode_pool_meta_replica(pool_meta_persisted_v1_replica_test_data("pool-0")), + decode_pool_meta_replica(pool_meta_replica_test_data("pool-0")), + ]) + .expect("equivalent v1 and v2 replicas should converge"); + + assert_eq!(selection.meta.version, POOL_META_VERSION); + let data = selection + .meta + .encode_config_data_for_v2_gate(false) + .expect("the selected v2 floor must remain writable"); + assert_eq!(LittleEndian::read_u16(&data[2..4]), POOL_META_VERSION); + } + #[test] fn pool_meta_replica_selection_distinguishes_absent_from_unrecoverable() { assert!(matches!(decode_pool_meta_replica(Vec::new()), PoolMetaReplica::Corrupt(_))); @@ -8650,7 +9404,7 @@ mod tests { .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")); + assert!(err.to_string().contains("version 2 tuple has unsupported field count")); } #[test] @@ -8990,6 +9744,17 @@ mod tests { #[test] fn pool_meta_persists_decommission_resume_queues() { let start_time = OffsetDateTime::now_utc(); + let unresolved_entry = DecommissionUnresolvedEntry { + bucket: "bucket-b".to_string(), + object: "prefix/unresolved.txt".to_string(), + pool_index: 0, + set_index: 1, + source_generation: start_time, + candidate_count: 2, + disk_error_count: 1, + observed_at: start_time, + reason: "metadata_resolution_failed".to_string(), + }; let pool_meta = PoolMeta { version: POOL_META_VERSION, pools: vec![PoolStatus { @@ -9010,6 +9775,7 @@ mod tests { bytes_failed: 128, terminal_reload_attempt_at: Some(start_time), terminal_reload_failures: vec!["complete_decommission: peer node-a failed".to_string()], + unresolved_entries: vec![unresolved_entry.clone()], ..Default::default() }), }], @@ -9049,6 +9815,7 @@ mod tests { restored_decommission.terminal_reload_failures, vec!["complete_decommission: peer node-a failed".to_string()] ); + assert_eq!(restored_decommission.unresolved_entries, vec![unresolved_entry]); assert!(restored_decommission.queued); assert_eq!(restored_decommission.items_since_last_progress_save(), 0); } @@ -9093,7 +9860,7 @@ mod tests { fn pool_meta_decode_supports_legacy_payload() { let start_time = OffsetDateTime::now_utc(); let legacy_meta = LegacyPoolMeta { - version: POOL_META_VERSION, + version: POOL_META_V1_VERSION, pools: vec![LegacyPoolStatus { id: 3, cmd_line: "/legacy/pool".to_string(), @@ -9119,8 +9886,9 @@ mod tests { let persisted_decode: std::result::Result = rmp_serde::from_slice(&legacy_payload); assert!(persisted_decode.is_err()); - let decoded = PoolMeta::decode_pool_meta_payload(&legacy_payload).expect("legacy payload should decode"); - assert_eq!(decoded.version, POOL_META_VERSION); + let decoded = + PoolMeta::decode_pool_meta_payload(POOL_META_V1_VERSION, &legacy_payload).expect("legacy payload should decode"); + assert_eq!(decoded.version, POOL_META_V1_VERSION); assert!(!decoded.dont_save, "runtime-only flag should reset on load"); assert_eq!(decoded.pools.len(), 1); assert_eq!(decoded.pools[0].id, 3); @@ -9140,6 +9908,7 @@ mod tests { assert!(decommission.bucket.is_empty()); assert!(decommission.prefix.is_empty()); assert!(decommission.object.is_empty()); + assert!(decommission.unresolved_entries.is_empty()); } #[test] @@ -9153,17 +9922,17 @@ mod tests { } let payload = rmp_serde::to_vec_named(&LegacyPoolMetaWithUnknownField { - version: POOL_META_VERSION, + version: POOL_META_V1_VERSION, pools: Vec::new(), dont_save: true, unexpected: true, }) .expect("legacy pool metadata with unknown field should serialize"); - let err = PoolMeta::decode_pool_meta_payload(payload.as_slice()) + let err = PoolMeta::decode_pool_meta_payload(POOL_META_V1_VERSION, payload.as_slice()) .expect_err("unknown legacy pool metadata field should fail decode"); let rendered = err.to_string(); - assert!(rendered.contains("PoolMeta decode failed for both persisted and legacy formats")); + assert!(rendered.contains("PoolMeta v1 decode failed for both persisted and legacy formats")); assert!(rendered.contains("unknown field") || rendered.contains("missing field")); } @@ -9183,10 +9952,10 @@ mod tests { }) .expect("pool metadata with unknown field should serialize"); - let err = PoolMeta::decode_pool_meta_payload(payload.as_slice()) + let err = PoolMeta::decode_pool_meta_payload(POOL_META_VERSION, payload.as_slice()) .expect_err("unknown persisted pool metadata field should fail decode"); let rendered = err.to_string(); - assert!(rendered.contains("PoolMeta decode failed for both persisted and legacy formats")); + assert!(rendered.contains("PoolMeta v2 decode failed")); assert!(rendered.contains("unknown field") || rendered.contains("missing field")); } @@ -9202,12 +9971,9 @@ mod tests { }) .expect("pool metadata without pools should serialize"); - let err = PoolMeta::decode_pool_meta_payload(payload.as_slice()) + let err = PoolMeta::decode_pool_meta_payload(POOL_META_VERSION, payload.as_slice()) .expect_err("missing persisted pool metadata pools should fail decode"); - assert!( - err.to_string() - .contains("PoolMeta decode failed for both persisted and legacy formats") - ); + assert!(err.to_string().contains("PoolMeta v2 decode failed")); } #[test] @@ -9298,12 +10064,9 @@ mod tests { }) .expect("pool metadata with unknown decommission field should serialize"); - let err = PoolMeta::decode_pool_meta_payload(payload.as_slice()) + let err = PoolMeta::decode_pool_meta_payload(POOL_META_VERSION, payload.as_slice()) .expect_err("unknown persisted decommission metadata field should fail decode"); - assert!( - err.to_string() - .contains("PoolMeta decode failed for both persisted and legacy formats") - ); + assert!(err.to_string().contains("PoolMeta v2 decode failed")); } #[test] @@ -9330,7 +10093,8 @@ mod tests { .serialize(&mut Serializer::new(&mut payload)) .expect("persisted payload should serialize"); - let err = PoolMeta::decode_pool_meta_payload(&payload).expect_err("invalid terminal state should fail decode"); + let err = PoolMeta::decode_pool_meta_payload(POOL_META_VERSION, &payload) + .expect_err("invalid terminal state should fail decode"); assert!(err.to_string().contains("invalid decommission terminal state")); } @@ -9338,7 +10102,7 @@ mod tests { fn pool_meta_decode_rejects_invalid_legacy_decommission_terminal_state() { let start_time = OffsetDateTime::now_utc(); let legacy_meta = LegacyPoolMeta { - version: POOL_META_VERSION, + version: POOL_META_V1_VERSION, pools: vec![LegacyPoolStatus { id: 1, cmd_line: "/legacy/pool".to_string(), @@ -9359,7 +10123,8 @@ mod tests { .serialize(&mut Serializer::new(&mut payload)) .expect("legacy payload should serialize"); - let err = PoolMeta::decode_pool_meta_payload(&payload).expect_err("invalid legacy terminal state should fail decode"); + let err = PoolMeta::decode_pool_meta_payload(POOL_META_V1_VERSION, &payload) + .expect_err("invalid legacy terminal state should fail decode"); assert!(err.to_string().contains("invalid decommission terminal state")); } } @@ -9429,31 +10194,80 @@ async fn record_decommission_entry_error( entry_error: &Arc>>, rx: &CancellationToken, err: Error, -) { +) -> bool { if rx.is_cancelled() { - return; + return false; } let mut first_err = entry_error.lock().await; if first_err.is_none() && !rx.is_cancelled() { *first_err = Some(err); rx.cancel(); + return true; } + false +} + +fn ensure_decommission_unresolved_verification_disk_count( + expected: usize, + actual: usize, + pool_index: usize, + set_index: usize, +) -> Result<()> { + if actual == expected { + return Ok(()); + } + Err(Error::other(format!( + "decommission unresolved-entry verification for pool {pool_index} set {set_index} requires all {expected} source disks, but only {actual} are online" + ))) } impl SetDisks { - #[tracing::instrument(skip(self, rx, cb_func, entry_error))] + async fn decommission_unresolved_entry_absent_on_all_disks( + &self, + pool_index: usize, + entry: &DecommissionUnresolvedEntry, + ) -> Result { + let (disks, _) = self.get_online_disks_with_healing(false).await; + ensure_decommission_unresolved_verification_disk_count(self.set_drive_count, disks.len(), pool_index, entry.set_index)?; + + let object = encode_dir_object(&entry.object); + let reads = join_all(disks.iter().map(|disk| disk.read_xl(&entry.bucket, &object, false))).await; + let mut found = false; + for read in reads { + match read { + Ok(_) => found = true, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => {} + Err(err) => { + return Err(Error::other(format!( + "decommission unresolved-entry verification failed for pool {pool_index} set {} path {}/{}: {err}", + entry.set_index, entry.bucket, entry.object + ))); + } + } + } + Ok(!found) + } + + #[tracing::instrument(skip(self, store, rx, cb_func, entry_error))] + #[allow(clippy::too_many_arguments)] async fn list_objects_to_decommission( self: &Arc, + store: Arc, rx: CancellationToken, bucket_info: DecomBucketInfo, cb_func: ListCallback, entry_error: Arc>>, pool_index: usize, set_index: usize, + source_generation: OffsetDateTime, + require_all_disks: bool, ) -> Result<()> { let (disks, _) = self.get_online_disks_with_healing(false).await; ensure_decommission_listing_disks_available(!disks.is_empty(), &bucket_info.name)?; + if require_all_disks { + ensure_decommission_unresolved_verification_disk_count(self.set_drive_count, disks.len(), pool_index, set_index)?; + } let listing_quorum = self.set_drive_count.div_ceil(2); @@ -9471,6 +10285,8 @@ impl SetDisks { let unresolved_prefix = bucket_info.prefix.clone(); let unresolved_pool_index = pool_index; let unresolved_set_index = set_index; + let unresolved_generation = source_generation; + let unresolved_store = store; list_path_raw( rx, @@ -9490,8 +10306,10 @@ impl SetDisks { let prefix = unresolved_prefix.clone(); let unresolved_error = unresolved_error.clone(); let unresolved_rx = unresolved_rx.clone(); + let unresolved_store = unresolved_store.clone(); let pool_index = unresolved_pool_index; let set_index = unresolved_set_index; + let source_generation = unresolved_generation; let disk_error_count = errs.iter().flatten().count(); if unresolved_rx.is_cancelled() { return Box::pin(async {}); @@ -9505,6 +10323,7 @@ impl SetDisks { disk_error_count, pool_index, set_index, + source_generation, ) { Ok(entry) => { warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); @@ -9512,10 +10331,11 @@ impl SetDisks { cb_func(entry).await; }) } - Err(err) => Box::pin(async move { + Err(unresolved_entry) => Box::pin(async move { if unresolved_rx.is_cancelled() { return; } + let err = decommission_unresolved_listing_error(&unresolved_entry); warn!( event = EVENT_DECOMMISSION_BUCKET, component = LOG_COMPONENT_ECSTORE, @@ -9526,6 +10346,13 @@ impl SetDisks { error = %err, "Decommission listing failed closed on unresolved metadata" ); + let err = match unresolved_store + .persist_decommission_unresolved_entry(pool_index, source_generation, unresolved_entry) + .await + { + Ok(()) => err, + Err(ledger_err) => Error::other(format!("{err}; {ledger_err}")), + }; record_decommission_entry_error(&unresolved_error, &unresolved_rx, err).await; }), } @@ -9737,25 +10564,29 @@ mod pools_tests { DECOMMISSION_ENTRY_CONCURRENCY_HARD_CAP, DECOMMISSION_ENTRY_QUEUE_HARD_CAP, DECOMMISSION_META_PREFIXES, DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DECOMMISSION_SOURCE_CHANGED_EXHAUSTION_LIMIT, DecomBucketInfo, DecommissionCanceler, DecommissionDurableIlmReceipt, - DecommissionEntryEnqueueResult, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, POOL_META_NAME, - PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, PoolStatus, QueuedDecommissionEntry, REBAL_META_NAME, - acquire_pool_rebalance_activation_locks, apply_decommission_status_space_info, await_decommission_worker, - bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler, - clamp_decommission_entry_concurrency, classify_decommission_terminal_state, count_decommission_item, - decommission_cancel_signal_result, decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix, + DecommissionEntryEnqueueResult, DecommissionStartPoolState, DecommissionTerminalState, DecommissionUnresolvedEntry, + ListCallback, POOL_META_NAME, POOL_META_V1_VERSION, POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, + PoolStatus, QueuedDecommissionEntry, REBAL_META_NAME, acquire_pool_rebalance_activation_locks, + apply_decommission_status_space_info, await_decommission_worker, bind_decommission_cancelers, + bind_missing_decommission_cancelers, cancel_decommission_canceler, clamp_decommission_entry_concurrency, + classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result, + decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix, decommission_durable_ilm_receipt_run_token, decommission_entry_queue_capacity, decommission_item_size, - decommission_meta_bucket_options, decommission_retry_backoff_delay, decommission_start_pool_state, dedup_indices, - default_decommission_bucket_concurrency, default_decommission_entry_concurrency, drain_decommission_entry_queue, - enqueue_decommission_entry, ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, - ensure_decommission_generation, ensure_decommission_listing_disks_available, ensure_decommission_not_rebalancing, - ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool, ensure_decommission_start_local_leader, + decommission_meta_bucket_options, decommission_retry_backoff_delay, decommission_start_pool_state, + decommission_unresolved_listing_error, dedup_indices, default_decommission_bucket_concurrency, + default_decommission_entry_concurrency, drain_decommission_entry_queue, enqueue_decommission_entry, + ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_generation, + ensure_decommission_listing_disks_available, ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, + ensure_decommission_start_keeps_active_pool, ensure_decommission_start_local_leader, ensure_decommission_start_pool_states, ensure_decommission_start_rebalance_meta_allowed, ensure_decommission_start_target_capacity, ensure_decommission_terminal_operation_supported, - ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, get_by_index, guard_decommission_cancelers, - has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested, - load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done, - merge_decommission_durable_ilm_receipts, merge_pool_status_refresh, missing_decommission_worker_prefix, - observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, + ensure_decommission_unresolved_verification_disk_count, ensure_local_decommission_pool_leaders, + ensure_valid_decommission_pool_index, get_by_index, guard_decommission_cancelers, has_active_decommission_canceler, + is_decommission_active, is_decommission_cancel_requested, load_decommission_entry_versions, + local_decommission_queue_prefix, mark_decommission_bucket_done, merge_decommission_durable_ilm_receipts, + merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, + pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, + reconcile_decommission_unresolved_entries_for_completion, record_decommission_unresolved_entry, require_decommission_store, reserve_decommission_start_cancelers, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions, @@ -9781,13 +10612,16 @@ mod pools_tests { manual_transition_job::{ManualTransitionJobRecord, manual_transition_job_record_object_name}, validate_durable_ilm_record, }; + use crate::bucket::metadata_sys; use crate::data_movement; - use crate::disk::endpoint::Endpoint; + use crate::disk::{STORAGE_FORMAT_FILE, endpoint::Endpoint}; use crate::error::{Error, StorageError}; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::runtime::instance::InstanceContext; use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats}; + use crate::storage_api_contracts::bucket::MakeBucketOptions; use crate::store::ECStore; + use byteorder::{ByteOrder, LittleEndian}; use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo}; use rustfs_filemeta::{MetaCacheEntries, MetadataResolutionParams}; use rustfs_rio::Index; @@ -10133,6 +10967,33 @@ mod pools_tests { assert_eq!(info.bytes_done, 1_024); } + #[test] + fn test_merge_pool_status_refresh_preserves_observed_v2_floor() { + let timestamp = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + version: POOL_META_V1_VERSION, + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: timestamp, + decommission: Some(PoolDecommissionInfo::default()), + }], + dont_save: false, + }; + let persisted = PoolMeta { + version: POOL_META_VERSION, + pools: current.pools.clone(), + dont_save: false, + }; + + assert!(!merge_pool_status_refresh(&mut current, persisted, &[true])); + assert_eq!(current.version, POOL_META_VERSION); + let encoded = current + .encode_config_data_for_v2_gate(false) + .expect("a peer-observed v2 floor must remain sticky"); + assert_eq!(LittleEndian::read_u16(&encoded[2..4]), POOL_META_VERSION); + } + #[test] fn test_merge_pool_status_refresh_keeps_newer_local_active_over_older_terminal() { let older = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); @@ -11348,7 +12209,8 @@ mod pools_tests { #[test] fn test_resolve_decommission_partial_listing_entry_rejects_unresolved_metadata() { - let err = resolve_decommission_partial_listing_entry( + let generation = OffsetDateTime::now_utc(); + let unresolved_entry = resolve_decommission_partial_listing_entry( MetaCacheEntries(vec![None]), MetadataResolutionParams { dir_quorum: 2, @@ -11361,23 +12223,232 @@ mod pools_tests { 1, 2, 3, + generation, ) .expect_err("unresolved partial listing must fail closed"); - let message = err.to_string(); + assert_eq!(unresolved_entry.bucket, "bucket-a"); + assert_eq!(unresolved_entry.object, "prefix/"); + assert_eq!(unresolved_entry.pool_index, 2); + assert_eq!(unresolved_entry.set_index, 3); + assert_eq!(unresolved_entry.source_generation, generation); + assert_eq!(unresolved_entry.candidate_count, 0); + assert_eq!(unresolved_entry.disk_error_count, 1); + assert_eq!(unresolved_entry.reason, "metadata_resolution_failed"); + + let message = decommission_unresolved_listing_error(&unresolved_entry).to_string(); assert!(message.contains("decommission listing could not resolve metadata")); assert!(message.contains("bucket-a/prefix/")); assert!(message.contains("pool 2 set 3")); assert!(message.contains("1 disk error(s)")); } + #[test] + fn unresolved_entry_ledger_requires_individual_verification_before_completion() { + let generation = OffsetDateTime::now_utc(); + let mut pool_meta = PoolMeta { + version: POOL_META_VERSION, + pools: vec![PoolStatus { + id: 0, + cmd_line: "/data/pool".to_string(), + last_update: generation, + decommission: Some(PoolDecommissionInfo { + start_time: Some(generation), + ..Default::default() + }), + }], + dont_save: true, + }; + let unresolved_entry = DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: generation, + candidate_count: 1, + disk_error_count: 0, + observed_at: generation, + reason: "metadata_resolution_failed".to_string(), + }; + + assert!( + record_decommission_unresolved_entry(&mut pool_meta, 0, generation, unresolved_entry.clone()) + .expect("active generation should accept unresolved entry") + ); + let err = reconcile_decommission_unresolved_entries_for_completion(&mut pool_meta, 0, None, None) + .expect_err("unverified completion must retain unresolved entries"); + assert!(err.to_string().contains("1 unresolved listing entries remain")); + assert_eq!( + pool_meta.pools[0] + .decommission + .as_ref() + .expect("decommission should exist") + .unresolved_entries + .len(), + 1 + ); + + let err = reconcile_decommission_unresolved_entries_for_completion(&mut pool_meta, 0, Some(generation), None) + .expect_err("a same-generation sweep without entry proof must retain the ledger"); + assert!(err.to_string().contains("not individually verified")); + + let stale_verified = vec![unresolved_entry.clone()]; + let mut replacement = unresolved_entry; + replacement.disk_error_count = 2; + replacement.observed_at = generation + Duration::seconds(1); + assert!( + record_decommission_unresolved_entry(&mut pool_meta, 0, generation, replacement.clone()) + .expect("a newer observation should replace the ledger entry") + ); + let err = + reconcile_decommission_unresolved_entries_for_completion(&mut pool_meta, 0, Some(generation), Some(&stale_verified)) + .expect_err("stale entry verification must not clear a concurrent replacement"); + assert!(err.to_string().contains("not individually verified")); + + let verified = vec![replacement]; + reconcile_decommission_unresolved_entries_for_completion(&mut pool_meta, 0, Some(generation), Some(&verified)) + .expect("individually verified entries should reconcile"); + assert!( + pool_meta.pools[0] + .decommission + .as_ref() + .expect("decommission should exist") + .unresolved_entries + .is_empty() + ); + } + + #[tokio::test] + async fn final_sweep_persists_unresolved_entry_from_real_listing() { + let (dirs, store) = metadata_sys::test_support::isolated_store_over_temp_disks().await; + let bucket = "decommission-final-sweep-unresolved"; + let object = "corrupt-object"; + store + .peer_sys + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("test bucket should be created"); + metadata_sys::init_bucket_metadata_sys(store.clone(), vec![bucket.to_string()]).await; + + let generation = OffsetDateTime::now_utc(); + { + let mut pool_meta = store.pool_meta.write().await; + pool_meta.dont_save = false; + pool_meta.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(generation), + ..Default::default() + }); + } + + for (disk_index, dir) in dirs.iter().enumerate() { + let object_dir = dir.path().join(bucket).join(object); + tokio::fs::create_dir_all(&object_dir) + .await + .expect("corrupt object directory should be created"); + tokio::fs::write(object_dir.join(STORAGE_FORMAT_FILE), format!("corrupt-xl-meta-{disk_index}").into_bytes()) + .await + .expect("divergent corrupt metadata should be written"); + } + + let err = store + .check_after_decommission(0, &CancellationToken::new(), generation) + .await + .expect_err("real final sweep must fail closed on unresolved listing metadata"); + assert!(err.to_string().contains("decommission listing could not resolve metadata")); + + let in_memory_entries = store.pool_meta.read().await.pools[0] + .decommission + .as_ref() + .expect("decommission should remain active") + .unresolved_entries + .clone(); + assert_eq!(in_memory_entries.len(), 1); + let unresolved_entry = &in_memory_entries[0]; + assert_eq!(unresolved_entry.bucket, bucket); + assert_eq!(unresolved_entry.object, object); + assert_eq!(unresolved_entry.pool_index, 0); + assert_eq!(unresolved_entry.set_index, 0); + assert_eq!(unresolved_entry.source_generation, generation); + assert_eq!(unresolved_entry.candidate_count, 4); + assert_eq!(unresolved_entry.disk_error_count, 0); + assert_eq!(unresolved_entry.reason, "metadata_resolution_failed"); + + let mut restored = PoolMeta::default(); + restored + .load_no_lock_from_replicas(vec![store.pools[0].clone()]) + .await + .expect("persisted pool metadata should reload after the final-sweep failure"); + assert_eq!( + restored.pools[0] + .decommission + .as_ref() + .expect("reloaded decommission should exist") + .unresolved_entries, + in_memory_entries + ); + } + + #[tokio::test] + async fn unresolved_entry_probe_verifies_absence_on_every_source_disk() { + let (_dirs, store) = metadata_sys::test_support::isolated_store_over_temp_disks().await; + let bucket = "decommission-final-sweep-absent-ledger"; + store + .peer_sys + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("test bucket should be created"); + metadata_sys::init_bucket_metadata_sys(store.clone(), vec![bucket.to_string()]).await; + + let generation = OffsetDateTime::now_utc(); + let unresolved_entry = DecommissionUnresolvedEntry { + bucket: bucket.to_string(), + object: "absent-object".to_string(), + pool_index: 0, + set_index: 0, + source_generation: generation, + candidate_count: 1, + disk_error_count: 1, + observed_at: generation, + reason: "metadata_resolution_failed".to_string(), + }; + { + let mut pool_meta = store.pool_meta.write().await; + pool_meta.version = POOL_META_VERSION; + pool_meta.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(generation), + unresolved_entries: vec![unresolved_entry.clone()], + ..Default::default() + }); + } + + assert!( + store.pools[0].disk_set[0] + .decommission_unresolved_entry_absent_on_all_disks(0, &unresolved_entry) + .await + .expect("all source disks should be readable") + ); + let verified = vec![unresolved_entry.clone()]; + + let mut pool_meta = store.pool_meta.write().await; + reconcile_decommission_unresolved_entries_for_completion(&mut pool_meta, 0, Some(generation), Some(&verified)) + .expect("the individually verified entry should reconcile"); + assert!( + pool_meta.pools[0] + .decommission + .as_ref() + .expect("decommission should remain present") + .unresolved_entries + .is_empty() + ); + } + #[tokio::test] async fn test_record_decommission_entry_error_cancels_listing_and_preserves_first_error() { let entry_error = Arc::new(tokio::sync::Mutex::new(None)); let rx = CancellationToken::new(); - record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await; - record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await; + assert!(record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await); + assert!(!record_decommission_entry_error(&entry_error, &rx, Error::OperationCanceled).await); assert!(rx.is_cancelled()); assert!(matches!(*entry_error.lock().await, Some(Error::SlowDown))); @@ -11389,11 +12460,20 @@ mod pools_tests { let rx = CancellationToken::new(); rx.cancel(); - record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await; + assert!(!record_decommission_entry_error(&entry_error, &rx, Error::SlowDown).await); assert!(entry_error.lock().await.is_none()); } + #[test] + fn unresolved_entry_verification_requires_every_source_disk() { + assert!(ensure_decommission_unresolved_verification_disk_count(4, 4, 0, 1).is_ok()); + + let err = ensure_decommission_unresolved_verification_disk_count(4, 3, 0, 1) + .expect_err("an offline source disk must block ledger reconciliation"); + assert!(err.to_string().contains("requires all 4 source disks, but only 3 are online")); + } + #[test] fn test_resolve_decommission_listing_error_preserves_real_listing_failure() { let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled)) @@ -12149,6 +13229,11 @@ mod pools_tests { assert!(ensure_decommission_start_allowed(DecommissionStartPoolState::Active).is_ok()); } + #[test] + fn test_ensure_decommission_start_allowed_allows_retryable_state() { + assert!(ensure_decommission_start_allowed(DecommissionStartPoolState::Retryable).is_ok()); + } + #[test] fn test_decommission_start_pool_state_reports_missing_pool() { assert_eq!(decommission_start_pool_state(None), DecommissionStartPoolState::Missing); @@ -12216,6 +13301,32 @@ mod pools_tests { assert_eq!(decommission_start_pool_state(Some(&pool)), DecommissionStartPoolState::Blocked); } + #[test] + fn test_decommission_start_pool_state_reports_terminal_pool_with_unresolved_entries_as_retryable() { + let pool = PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + failed: true, + unresolved_entries: vec![DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: OffsetDateTime::UNIX_EPOCH, + candidate_count: 1, + disk_error_count: 1, + observed_at: OffsetDateTime::UNIX_EPOCH, + reason: "metadata_resolution_failed".to_string(), + }], + ..Default::default() + }), + }; + + assert_eq!(decommission_start_pool_state(Some(&pool)), DecommissionStartPoolState::Retryable); + } + #[test] fn test_decommission_start_pool_state_reports_completed_pool() { let pool = PoolStatus { @@ -12408,6 +13519,43 @@ mod pools_tests { assert!(ensure_decommission_start_pool_states(&meta, &[0]).is_ok()); } + #[test] + fn test_ensure_decommission_start_pool_states_allows_retryable_pool_with_active_peer() { + let meta = PoolMeta { + pools: vec![ + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + failed: true, + unresolved_entries: vec![DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: OffsetDateTime::UNIX_EPOCH, + candidate_count: 1, + disk_error_count: 1, + observed_at: OffsetDateTime::UNIX_EPOCH, + reason: "metadata_resolution_failed".to_string(), + }], + ..Default::default() + }), + }, + PoolStatus { + id: 1, + cmd_line: "pool-1".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }, + ], + ..Default::default() + }; + + assert!(ensure_decommission_start_pool_states(&meta, &[0]).is_ok()); + } + #[test] fn test_ensure_valid_decommission_pool_index_accepts_in_range_index() { assert!(ensure_valid_decommission_pool_index(4, 3).is_ok()); @@ -12620,21 +13768,29 @@ mod pools_tests { #[test] fn test_ensure_decommission_clear_allowed_allows_failed_or_canceled() { - assert!(ensure_decommission_clear_allowed(true, true, false, true, false).is_ok()); - assert!(ensure_decommission_clear_allowed(true, true, false, false, true).is_ok()); + assert!(ensure_decommission_clear_allowed(true, true, false, true, false, 0).is_ok()); + assert!(ensure_decommission_clear_allowed(true, true, false, false, true, 0).is_ok()); } #[test] fn test_ensure_decommission_clear_allowed_rejects_active_or_completed() { - let active = ensure_decommission_clear_allowed(true, true, false, false, false) + let active = ensure_decommission_clear_allowed(true, true, false, false, false, 0) .expect_err("active decommission should not be clearable"); assert!(matches!(active, Error::DecommissionAlreadyRunning)); - let complete = ensure_decommission_clear_allowed(true, true, true, false, false) + let complete = ensure_decommission_clear_allowed(true, true, true, false, false, 0) .expect_err("completed decommission should not be clearable"); assert!(matches!(complete, Error::DecommissionNotStarted)); } + #[test] + fn test_ensure_decommission_clear_allowed_rejects_unresolved_entries() { + let err = ensure_decommission_clear_allowed(true, true, false, true, false, 1) + .expect_err("unresolved entries must survive until a retry reconciles them"); + + assert!(err.to_string().contains("must be reconciled by retrying decommission")); + } + #[test] fn test_pool_meta_clear_decommission_restores_failed_or_canceled_pool() { for decommission in [ @@ -12664,6 +13820,49 @@ mod pools_tests { } } + #[test] + fn test_pool_meta_clear_decommission_preserves_unresolved_entries_for_retry() { + let generation = OffsetDateTime::UNIX_EPOCH; + let unresolved_entry = DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: generation, + candidate_count: 1, + disk_error_count: 1, + observed_at: generation, + reason: "metadata_resolution_failed".to_string(), + }; + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: generation, + decommission: Some(PoolDecommissionInfo { + failed: true, + unresolved_entries: vec![unresolved_entry.clone()], + ..Default::default() + }), + }], + ..Default::default() + }; + + let err = meta + .clear_decommission(0) + .expect_err("clear must not discard the unresolved-entry recovery ledger"); + + assert!(err.to_string().contains("must be reconciled by retrying decommission")); + assert_eq!( + meta.pools[0] + .decommission + .as_ref() + .expect("failed state should remain retryable") + .unresolved_entries, + vec![unresolved_entry] + ); + } + #[test] fn test_pool_meta_clear_decommission_rejects_active_or_completed_pool() { for decommission in [ @@ -12732,6 +13931,108 @@ mod pools_tests { assert!(validate_start_decommission_request(&[0], false).is_ok()); } + #[test] + fn test_decommission_retry_preserves_and_rebinds_unresolved_entries() { + let previous_generation = OffsetDateTime::UNIX_EPOCH; + let unresolved_entry = DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: previous_generation, + candidate_count: 1, + disk_error_count: 1, + observed_at: previous_generation, + reason: "metadata_resolution_failed".to_string(), + }; + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: previous_generation, + decommission: Some(PoolDecommissionInfo { + failed: true, + decommissioned_buckets: vec!["bucket-a".to_string()], + items_decommissioned: 7, + bytes_done: 1024, + unresolved_entries: vec![unresolved_entry], + ..Default::default() + }), + }], + ..Default::default() + }; + + meta.decommission( + 0, + PoolSpaceInfo { + total: 200, + free: 50, + used: 150, + }, + ) + .expect("failed decommission with unresolved entries should retry atomically"); + + let info = meta.pools[0] + .decommission + .as_ref() + .expect("retried decommission metadata should exist"); + let next_generation = info.start_time.expect("retry should assign a new generation"); + assert!(next_generation > previous_generation); + assert_eq!(info.decommissioned_buckets, vec!["bucket-a".to_string()]); + assert_eq!(info.items_decommissioned, 7); + assert_eq!(info.bytes_done, 1024); + assert_eq!(info.unresolved_entries.len(), 1); + assert_eq!(info.unresolved_entries[0].source_generation, next_generation); + } + + #[test] + fn test_queued_decommission_retry_rebinds_unresolved_entries_when_promoted() { + let previous_generation = OffsetDateTime::UNIX_EPOCH; + let mut meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: previous_generation, + decommission: Some(PoolDecommissionInfo { + canceled: true, + unresolved_entries: vec![DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + pool_index: 0, + set_index: 0, + source_generation: previous_generation, + candidate_count: 1, + disk_error_count: 1, + observed_at: previous_generation, + reason: "metadata_resolution_failed".to_string(), + }], + ..Default::default() + }), + }], + ..Default::default() + }; + + meta.queue_decommission( + 0, + PoolSpaceInfo { + total: 100, + free: 25, + used: 75, + }, + ) + .expect("canceled decommission with unresolved entries should queue an atomic retry"); + + assert!(meta.is_suspended(0)); + assert!(meta.promote_queued_decommission(0)); + let promoted = meta.pools[0] + .decommission + .as_ref() + .expect("promoted decommission metadata should exist"); + let generation = promoted.start_time.expect("promotion should assign a generation"); + assert_eq!(promoted.unresolved_entries.len(), 1); + assert_eq!(promoted.unresolved_entries[0].source_generation, generation); + } + #[test] fn test_pool_meta_queued_decommission_is_not_suspended_until_promoted() { let mut meta = PoolMeta { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 0ec7be855..6cd1452ae 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -637,7 +637,7 @@ mod tests { }; use crate::{ bucket::replication::{ReplicationState, ReplicationStatusType, replication_statuses_map}, - core::pools::{POOL_META_FORMAT, POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus}, + core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus}, disk::endpoint::Endpoint, error::{Error, Result, StorageError}, io_support::rio::{WritePlan, compression_metadata_value}, @@ -651,7 +651,6 @@ mod tests { range::HTTPRangeSpec, }, }; - use byteorder::{LittleEndian, WriteBytesExt}; #[cfg(feature = "test-util")] use futures::{StreamExt as _, TryStreamExt as _}; use http::HeaderMap; @@ -681,13 +680,7 @@ mod tests { 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 + meta.encode_config_data_for_test().expect("pool metadata should encode") } #[derive(Debug)] diff --git a/docs/operations/rolling-restart.md b/docs/operations/rolling-restart.md index 63b8bb9ee..fc263d432 100644 --- a/docs/operations/rolling-restart.md +++ b/docs/operations/rolling-restart.md @@ -31,6 +31,15 @@ failure pattern reported in rustfs/rustfs#4304. > older build is not supported: older readers ignore the sidecar and can > report an object checksum in place of the requested part checksum. +> [!WARNING] +> Writing pool metadata version 2 remains inactive unless both +> `RUSTFS_POOL_META_V2_WRITE=true` and +> `RUSTFS_POOL_META_V2_FLEET_CONFIRMED=true`. Leave either setting disabled +> until every node that can read or write `pool.bin` supports version 2. Once a node +> observes or writes version 2 it will not downgrade the file, and older +> binaries or rollback builds cannot read it. Unresolved decommission entries +> fail closed instead of being written in the version 1 format. + ## TL;DR - **Rolling restart (no downtime):** restart **one node at a time**, and wait diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 877296f29..5660ec2d9 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -16,7 +16,8 @@ use super::storage_api::admin_usecase::admin::get_server_info; use super::storage_api::admin_usecase::capacity::{ - PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity, get_total_usable_capacity_free, + DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity, + get_total_usable_capacity_free, }; use super::storage_api::admin_usecase::contract::StorageAdminApi; use super::storage_api::admin_usecase::contract::bucket::{BucketOperations as _, BucketOptions}; @@ -107,6 +108,8 @@ pub struct AdminPoolDecommissionInfo { pub bytes_failed: usize, #[serde(rename = "waitingReason")] pub waiting_reason: Option, + #[serde(rename = "unresolvedEntries", skip_serializing_if = "Vec::is_empty")] + pub unresolved_entries: Vec, } #[derive(Debug, Clone, serde::Serialize)] @@ -619,6 +622,7 @@ impl DefaultAdminUsecase { bytes_done: info.bytes_done, bytes_failed: info.bytes_failed, waiting_reason, + unresolved_entries: info.unresolved_entries, } } @@ -676,7 +680,7 @@ impl DefaultAdminUsecase { #[cfg(test)] mod tests { - use super::super::storage_api::admin_usecase::capacity::{PoolDecommissionInfo, PoolStatus}; + use super::super::storage_api::admin_usecase::capacity::{DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus}; use super::*; use time::OffsetDateTime; use tracing_subscriber::{Layer, Registry, layer::Context, prelude::*}; @@ -987,6 +991,17 @@ mod tests { items_decommission_failed: 1, bytes_done: 1024, bytes_failed: 64, + unresolved_entries: vec![DecommissionUnresolvedEntry { + bucket: "bucket-a".to_string(), + object: "prefix/unresolved.txt".to_string(), + pool_index: 3, + set_index: 1, + source_generation: OffsetDateTime::UNIX_EPOCH, + candidate_count: 2, + disk_error_count: 1, + observed_at: OffsetDateTime::UNIX_EPOCH, + reason: "metadata_resolution_failed".to_string(), + }], ..Default::default() }), }, @@ -1010,6 +1025,13 @@ mod tests { assert_eq!(value["decommissionInfo"]["objectsDecommissionedFailed"], 1); assert_eq!(value["decommissionInfo"]["bytesDecommissioned"], 1024); assert_eq!(value["decommissionInfo"]["bytesDecommissionedFailed"], 64); + assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["bucket"], "bucket-a"); + assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["object"], "prefix/unresolved.txt"); + assert_eq!( + value["decommissionInfo"]["unresolvedEntries"][0]["sourceGeneration"], + "1970-01-01T00:00:00Z" + ); + assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["reason"], "metadata_resolution_failed"); assert_eq!(value["decommissionInfo"]["waitingReason"], "queued"); } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index a08877d79..c31c18f36 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -31,6 +31,7 @@ pub(crate) mod admin { } pub(crate) mod capacity { + pub(crate) type DecommissionUnresolvedEntry = crate::storage::storage_api::ecstore_capacity::DecommissionUnresolvedEntry; pub(crate) type PoolDecommissionInfo = crate::storage::storage_api::ecstore_capacity::PoolDecommissionInfo; pub(crate) type PoolStatus = crate::storage::storage_api::ecstore_capacity::PoolStatus; pub(crate) type RebalStatus = crate::storage::storage_api::ecstore_rebalance::RebalStatus; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index dfdcc7300..9a99b52ec 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -399,7 +399,7 @@ pub(crate) mod ecstore_bucket { pub(crate) mod ecstore_capacity { pub(crate) use rustfs_ecstore::api::capacity::{ - PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free, + DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free, is_reserved_or_invalid_bucket, }; }