Merge remote-tracking branch 'origin/main' into overtrue/fix-1917-peer-reload-generation-merge

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

@@ -865,16 +865,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;
@@ -884,11 +890,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<()> {
@@ -5707,6 +5716,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]);

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<bool> {
+        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::<Vec<_>>()
+        };
+
+        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<PoolDecommissionInfo>, 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();
+    }
 }

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()),