From 4ad27860e3f74796ac7e21615f58acb9defe03c9 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 16:28:55 +0800 Subject: [PATCH] fix(ecstore): merge peer pool meta reload monotonically The peer reload_pool_meta handler blindly replaced in-memory pool metadata with the persisted snapshot, so a delayed or out-of-order reload could roll back newer local queued/canceled/failed/complete decommission state, and a missing pool.bin wiped local state to an empty default. Route peer reload through the same monotonic merge used by the admin status refresh (merge_pool_status_refresh): entries are replaced only when strictly newer and no local worker is active; missing snapshots fail closed. The helper now reports whether any entry was replaced or appended, and rejected stale/missing reloads are logged. The RPC handler spawns missing decommission workers only after a reload actually merged newer state, so duplicate deliveries cannot start workers for an older generation. Fixes rustfs/backlog#1917 --- crates/ecstore/src/core/pools.rs | 64 ++++- crates/ecstore/src/store/rebalance.rs | 327 ++++++++++++++++++++++++- rustfs/src/storage/rpc/node_service.rs | 8 +- 3 files changed, 389 insertions(+), 10 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index fb983dc0c..20b2c74ef 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -810,16 +810,22 @@ fn should_replace_pool_status_for_status_refresh( !has_active_worker && persisted.last_update > current.last_update } -fn merge_pool_status_refresh(current: &mut PoolMeta, persisted: PoolMeta, active_workers: &[bool]) { +/// Merges a persisted pool metadata snapshot into `current` monotonically: +/// a pool entry is replaced only when no active worker covers it and the +/// snapshot is strictly newer, so delayed snapshots never roll back local +/// 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 { if persisted.pools.is_empty() { - return; + return false; } if current.pools.is_empty() { *current = persisted; - return; + return true; } + let mut merged_newer = false; for (idx, persisted_pool) in persisted.pools.into_iter().enumerate() { if persisted_pool.id != idx { continue; @@ -829,11 +835,14 @@ fn merge_pool_status_refresh(current: &mut PoolMeta, persisted: PoolMeta, active if idx < current.pools.len() { if should_replace_pool_status_for_status_refresh(current.pools.get(idx), &persisted_pool, has_active_worker) { current.pools[idx] = persisted_pool; + merged_newer = true; } } else if idx == current.pools.len() && !has_active_worker { current.pools.push(persisted_pool); + merged_newer = true; } } + merged_newer } fn resolve_start_decommission_pool_meta_reload_result(result: Result<()>) -> Result<()> { @@ -5511,6 +5520,55 @@ mod pools_tests { assert_eq!(info.bytes_done, 1_024); } + #[test] + fn test_merge_pool_status_refresh_fails_closed_on_missing_persisted_pools() { + let newer = OffsetDateTime::from_unix_timestamp(2_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + pools: vec![decommission_test_pool_status( + 0, + Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + )], + ..Default::default() + }; + current.pools[0].last_update = newer; + + assert!( + !merge_pool_status_refresh(&mut current, PoolMeta::default(), &[false]), + "an empty persisted snapshot must fail closed instead of replacing local state" + ); + + let info = current.pools[0] + .decommission + .as_ref() + .expect("local decommission info should survive a missing snapshot"); + assert!(info.complete); + assert_eq!(current.pools[0].last_update, newer); + } + + #[test] + fn test_merge_pool_status_refresh_ignores_mislabeled_pool_entries() { + let older = OffsetDateTime::from_unix_timestamp(1_000).expect("test timestamp should be valid"); + let mut current = PoolMeta { + pools: vec![decommission_test_pool_status(0, None)], + ..Default::default() + }; + let mut persisted = PoolMeta { + pools: vec![decommission_test_pool_status(0, Some(PoolDecommissionInfo::default()))], + ..Default::default() + }; + persisted.pools[0].id = 7; + persisted.pools[0].last_update = older; + + assert!( + !merge_pool_status_refresh(&mut current, persisted, &[false]), + "a pool entry whose id does not match its index must be ignored" + ); + assert!(current.pools[0].decommission.is_none()); + } + #[test] fn test_dedup_indices_removes_duplicates_preserving_order() { assert_eq!(dedup_indices(&[0, 2, 1, 2, 3, 0]), vec![0, 2, 1, 3]); diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 7d7c1d91e..7447752bc 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -14,10 +14,15 @@ use super::*; use crate::config::storageclass; +use crate::core::pools::merge_pool_status_refresh; use crate::layout::pool_space::{ServerPoolsAvailableSpace, build_server_pools_available_space}; use crate::runtime::sources as runtime_sources; use crate::storage_api_contracts::{admin::StorageAdminApi, namespace::NamespaceLocking as _, object::ObjectOperations as _}; pub(in crate::store) mod support; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_POOLS: &str = "pools"; +const EVENT_POOL_META_RELOAD: &str = "pool_meta_reload"; use support::{ LatestObjectInfoCandidate, PoolErr, PoolObjInfo, RebalanceDeletePoolResult, pool_lookup_not_found_error, rebalance_disk_set_lookup_error, resolve_latest_object_info_candidates, resolve_rebalance_delete_from_all_pools_result, @@ -684,17 +689,49 @@ impl ECStore { ) } - pub async fn reload_pool_meta(&self) -> Result<()> { - let mut meta = PoolMeta::default(); + /// Peer reload entry: refreshes in-memory pool metadata from the shared + /// persisted snapshot. Returns whether newer state was actually merged so + /// callers only trigger missing-worker recovery after a real state change; + /// delayed snapshots are merged monotonically and never blind-assigned. + pub async fn reload_pool_meta(&self) -> Result { + let mut reloaded = PoolMeta::default(); resolve_store_rebalance_pool_meta_reload_result( - meta.load(self.pools[0].clone(), self.pools.clone()).await, + reloaded.load(self.pools[0].clone(), self.pools.clone()).await, "reload_pool_meta", )?; + // Lock order: release the decommission_cancelers guard before taking + // the pool_meta write guard; neither is held across the disk read. + let active_workers = { + let cancelers = self.decommission_cancelers.read().await; + cancelers.iter().map(Option::is_some).collect::>() + }; + + let incoming_has_pools = !reloaded.pools.is_empty(); let mut pool_meta = self.pool_meta.write().await; - *pool_meta = meta; - // *self.pool_meta.write().expect("operation should succeed") = meta; - Ok(()) + let merged_newer = merge_pool_status_refresh(&mut pool_meta, reloaded, &active_workers); + + if !merged_newer && !incoming_has_pools { + warn!( + event = EVENT_POOL_META_RELOAD, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + result = "ignored", + reason = "missing_metadata", + "Peer pool meta reload ignored because persisted metadata is missing" + ); + } else if !merged_newer { + debug!( + event = EVENT_POOL_META_RELOAD, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + result = "ignored", + reason = "stale_snapshot", + "Peer pool meta reload ignored as a stale snapshot" + ); + } + + Ok(merged_newer) } /// Disk information deduplication function @@ -860,6 +897,7 @@ fn lifecycle_delete_all_test_failure(phase: crate::object_api::LifecycleDeleteAl mod tests { use super::*; use crate::config::storageclass::{CLASS_RRS, CLASS_STANDARD, lookup_config_for_pools_without_env}; + use crate::core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolStatus}; use crate::disk::error::DiskError; use crate::layout::endpoint::Endpoint; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; @@ -870,6 +908,7 @@ mod tests { use rustfs_config::server_config::KVS; use rustfs_filemeta::FileInfo; use std::sync::Arc; + use time::{Duration as TimeDuration, OffsetDateTime}; use tokio_util::sync::CancellationToken; async fn setup_multi_pool_test_store( @@ -1712,4 +1751,280 @@ mod tests { .contains("failed to resolve rebalance disk set: pool index 2, set index 7, pool count 3") ); } + + fn reload_test_pool_status(decommission: Option, last_update: time::OffsetDateTime) -> PoolStatus { + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update, + decommission, + } + } + + fn reload_test_pool_meta(pool: PoolStatus) -> PoolMeta { + PoolMeta { + version: POOL_META_VERSION, + pools: vec![pool], + dont_save: false, + } + } + + async fn persist_reload_snapshot(store: &ECStore, snapshot: &PoolMeta) { + snapshot + .save(store.pools.clone()) + .await + .expect("pool meta snapshot should persist to every pool"); + } + + #[tokio::test] + #[serial_test::serial] + async fn peer_pool_meta_reload_does_not_rollback_newer_local_states() { + let (_temp_dir, store, shutdown) = setup_multi_pool_test_store("pool-meta-reload-stale", &[2]).await; + + let stale_time = OffsetDateTime::now_utc(); + let newer_time = stale_time + TimeDuration::seconds(30); + let progressed_states: [(&str, PoolDecommissionInfo); 4] = [ + ( + "queued", + PoolDecommissionInfo { + queued: true, + start_time: Some(stale_time), + ..Default::default() + }, + ), + ( + "canceled", + PoolDecommissionInfo { + canceled: true, + start_time: Some(stale_time), + ..Default::default() + }, + ), + ( + "failed", + PoolDecommissionInfo { + failed: true, + start_time: Some(stale_time), + ..Default::default() + }, + ), + ( + "complete", + PoolDecommissionInfo { + complete: true, + start_time: Some(stale_time), + ..Default::default() + }, + ), + ]; + + for (state_label, local_state) in progressed_states { + { + let mut pool_meta = store.pool_meta.write().await; + *pool_meta = reload_test_pool_meta(reload_test_pool_status(Some(local_state.clone()), newer_time)); + } + + // A delayed peer message carries a snapshot that predates the local progression. + let stale_snapshot = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + start_time: Some(stale_time), + ..Default::default() + }), + stale_time, + )); + persist_reload_snapshot(&store, &stale_snapshot).await; + + let merged_newer = store.reload_pool_meta().await.expect("stale reload should succeed"); + assert!( + !merged_newer, + "a delayed reload must not report merged newer state for the {state_label} progression" + ); + + let pool_meta = store.pool_meta.read().await; + let info = pool_meta.pools[0] + .decommission + .as_ref() + .expect("local decommission state should survive a stale reload"); + assert_eq!(info.queued, local_state.queued, "{state_label} queued flag must not roll back"); + assert_eq!(info.canceled, local_state.canceled, "{state_label} canceled flag must not roll back"); + assert_eq!(info.failed, local_state.failed, "{state_label} failed flag must not roll back"); + assert_eq!(info.complete, local_state.complete, "{state_label} complete flag must not roll back"); + assert_eq!( + pool_meta.pools[0].last_update, newer_time, + "{state_label} progress timestamp must be kept" + ); + } + + shutdown.cancel(); + } + + #[tokio::test] + #[serial_test::serial] + async fn peer_pool_meta_reload_merges_newer_state_and_is_idempotent_on_duplicate_delivery() { + let (_temp_dir, store, shutdown) = setup_multi_pool_test_store("pool-meta-reload-duplicate", &[2]).await; + + let older_time = OffsetDateTime::now_utc(); + let newer_time = older_time + TimeDuration::seconds(30); + + { + let mut pool_meta = store.pool_meta.write().await; + *pool_meta = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + items_decommissioned: 1, + ..Default::default() + }), + older_time, + )); + } + let newer_snapshot = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + complete: true, + items_decommissioned: 10, + ..Default::default() + }), + newer_time, + )); + persist_reload_snapshot(&store, &newer_snapshot).await; + + let merged_newer = store.reload_pool_meta().await.expect("first reload should succeed"); + assert!(merged_newer, "a strictly newer persisted snapshot must merge"); + + { + let pool_meta = store.pool_meta.read().await; + let info = pool_meta.pools[0].decommission.as_ref().expect("merged decommission state"); + assert!(info.complete); + assert_eq!(info.items_decommissioned, 10); + assert_eq!(pool_meta.pools[0].last_update, newer_time); + } + + // Redelivering the same generation must be a no-op. + let duplicate_merged = store.reload_pool_meta().await.expect("duplicate reload should succeed"); + assert!(!duplicate_merged, "a duplicate delivery must not re-apply merged state"); + + { + let pool_meta = store.pool_meta.read().await; + let info = pool_meta.pools[0].decommission.as_ref().expect("merged decommission state"); + assert!(info.complete); + assert_eq!(info.items_decommissioned, 10); + assert_eq!(pool_meta.pools[0].last_update, newer_time); + } + + shutdown.cancel(); + } + + #[tokio::test] + #[serial_test::serial] + async fn peer_pool_meta_reload_keeps_active_worker_progress_over_newer_snapshot() { + let (_temp_dir, store, shutdown) = setup_multi_pool_test_store("pool-meta-reload-worker", &[2]).await; + *store.decommission_cancelers.write().await = vec![Some(CancellationToken::new())]; + + let worker_time = OffsetDateTime::now_utc(); + let newer_time = worker_time + TimeDuration::seconds(30); + + { + let mut pool_meta = store.pool_meta.write().await; + *pool_meta = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + start_time: Some(worker_time), + items_decommissioned: 10, + bytes_done: 1_024, + ..Default::default() + }), + worker_time, + )); + } + // Even a strictly newer terminal snapshot must not override a live worker. + let newer_terminal_snapshot = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + newer_time, + )); + persist_reload_snapshot(&store, &newer_terminal_snapshot).await; + + let merged_newer = store + .reload_pool_meta() + .await + .expect("reload under an active worker should succeed"); + assert!(!merged_newer, "an active local worker must block snapshot replacement"); + + let pool_meta = store.pool_meta.read().await; + let info = pool_meta.pools[0] + .decommission + .as_ref() + .expect("worker progress should remain"); + assert!(!info.complete); + assert_eq!(info.items_decommissioned, 10); + assert_eq!(info.bytes_done, 1_024); + assert_eq!(pool_meta.pools[0].last_update, worker_time); + + shutdown.cancel(); + } + + #[tokio::test] + #[serial_test::serial] + async fn peer_pool_meta_reload_fails_closed_when_persisted_metadata_is_missing() { + let (temp_dir, store, shutdown) = setup_multi_pool_test_store("pool-meta-reload-missing", &[2]).await; + + let kept_time = OffsetDateTime::now_utc(); + { + let mut pool_meta = store.pool_meta.write().await; + *pool_meta = reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + kept_time, + )); + } + // Persist first so the test controls exactly what exists on disk. + persist_reload_snapshot( + &store, + &reload_test_pool_meta(reload_test_pool_status( + Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + kept_time, + )), + ) + .await; + + let mut deleted_any = false; + for disk_index in 0..2 { + let pool_bin_dir = temp_dir + .path() + .join(format!("pool0-disk{disk_index}")) + .join(crate::disk::RUSTFS_META_BUCKET) + .join(crate::core::pools::POOL_META_NAME); + if pool_bin_dir.exists() { + tokio::fs::remove_dir_all(&pool_bin_dir) + .await + .expect("persisted pool metadata object dir should be removable"); + deleted_any = true; + } + } + // The meta-bucket layout may nest objects per pool; fall back to removing + // every pool.bin object directory below the temp root. + if !deleted_any { + panic!("no pool.bin found under {:?}", temp_dir.path()); + } + + let merged_newer = store + .reload_pool_meta() + .await + .expect("reload with missing metadata should fail closed, not error"); + assert!(!merged_newer, "missing persisted metadata must not count as merged state"); + + let pool_meta = store.pool_meta.read().await; + let info = pool_meta.pools[0] + .decommission + .as_ref() + .expect("missing persisted metadata must not default local state away"); + assert!(info.complete); + assert_eq!(pool_meta.pools[0].last_update, kept_time); + + shutdown.cancel(); + } } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 2503e25c7..d8eaadcb4 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1987,8 +1987,10 @@ impl Node for NodeService { error_info: Some("errServerNotInitialized".to_string()), })); }; + // Recover missing workers only after the reload merged newer state; a + // stale or duplicate reload must not spawn workers for an older generation. match store.reload_pool_meta().await { - Ok(_) => match store.spawn_missing_local_decommission_routines().await { + Ok(true) => match store.spawn_missing_local_decommission_routines().await { Ok(_) => Ok(Response::new(ReloadPoolMetaResponse { success: true, error_info: None, @@ -1998,6 +2000,10 @@ impl Node for NodeService { error_info: Some(err.to_string()), })), }, + Ok(false) => Ok(Response::new(ReloadPoolMetaResponse { + success: true, + error_info: None, + })), Err(err) => Ok(Response::new(ReloadPoolMetaResponse { success: false, error_info: Some(err.to_string()),