fix(heal): scope replacement pool metadata to its owning set

crates/ecstore/src/store/heal.rs

@@ -183,6 +183,24 @@ impl ECStore {
         }
     }
 
+    /// Whether a replacement set owns this pool's `pool.bin` replica.
+    /// Placement is determined by the same hash as metadata writes, never by
+    /// whether a shard is currently readable on the replacement disk.
+    pub fn replacement_pool_metadata_required(&self, pool_index: usize, set_index: usize) -> Result<bool> {
+        let pool = self
+            .pools
+            .get(pool_index)
+            .ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?;
+        let target_set = pool.get_disks_for_heal_object(
+            POOL_META_NAME,
+            &HealOpts {
+                set: Some(set_index),
+                ..Default::default()
+            },
+        )?;
+        Ok(Arc::ptr_eq(&target_set, &pool.get_disks_by_key(POOL_META_NAME)))
+    }
+
     /// Return every live erasure set selected by an object-heal scope.
     pub async fn heal_erasure_set_scopes(&self, opts: &HealOpts) -> Result<Vec<(usize, usize)>> {
         let pools = self.get_pools_for_heal_object(opts)?;

crates/heal/src/heal/erasure_healer.rs

@@ -972,6 +972,12 @@ impl ErasureSetHealer {
             });
         }
 
+        // pool.bin is stored in one hashed set per pool. Requiring it on any
+        // other replacement set would retry a healthy absence forever.
+        if !self.storage.replacement_pool_metadata_required(&self.heal_opts)? {
+            return Ok(());
+        }
+
         let object_key = format!("{RUSTFS_META_BUCKET}/{POOL_META_NAME}");
         let checkpoint_key = compose_key(&object_key, None);
         let checkpoint = checkpoint_manager.get_checkpoint().await;
@@ -2040,6 +2046,7 @@ mod resume_loop_tests {
         Ok,
         /// The version vanished before heal ran (deleted mid-heal).
         VersionNotFound,
+        FileNotFound,
         /// A transient infrastructure condition (offline disk / unmet quorum):
         /// the version must be recorded as skipped and retried on a later pass.
         Transient,
@@ -2068,6 +2075,7 @@ mod resume_loop_tests {
         heal_calls: Mutex<Vec<(String, Option<String>)>>,
         list_include_lifecycle_object_info: Mutex<Vec<bool>>,
         replacement_target_identity_sequences: Mutex<VecDeque<Vec<ReplacementTargetIdentity>>>,
+        pool_metadata_placement: Mutex<Option<ReplacementCommitEvidence>>,
         fail_listing: AtomicBool,
         fail_listing_buckets: Mutex<HashSet<String>>,
     }
@@ -2164,6 +2172,7 @@ mod resume_loop_tests {
             let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
             match outcome {
                 HealOutcome::Ok => Ok((self.results.lock().unwrap().get(&key).cloned().unwrap_or_default(), None)),
+                HealOutcome::FileNotFound => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileNotFound)))),
                 HealOutcome::VersionNotFound => {
                     Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
                 }
@@ -2177,6 +2186,13 @@ mod resume_loop_tests {
         async fn heal_format(&self, _dry: bool) -> Result<(HealResultItem, Option<Error>)> {
             Ok((HealResultItem::default(), None))
         }
+        fn replacement_pool_metadata_required(&self, _opts: &HealOpts) -> Result<bool> {
+            match self.pool_metadata_placement.lock().unwrap().as_ref() {
+                Some(ReplacementCommitEvidence::Confirmed(required)) => Ok(*required),
+                Some(ReplacementCommitEvidence::Error(message)) => Err(Error::other(message.clone())),
+                None => Ok(true),
+            }
+        }
         async fn replacement_targets_have_version(
             &self,
             _bucket: &str,
@@ -2771,6 +2787,93 @@ mod resume_loop_tests {
         assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
     }
 
+    #[tokio::test]
+    async fn replacement_pool_metadata_placement_controls_completion() {
+        // Exercise both replacement intents and the admin/directory-backed path.
+        for automatic in [false, true] {
+            for placement in [Ok(false), Ok(true), Err("placement unavailable")] {
+                let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+                *env.storage.pool_metadata_placement.lock().unwrap() = Some(match placement {
+                    Ok(required) => ReplacementCommitEvidence::Confirmed(required),
+                    Err(message) => ReplacementCommitEvidence::Error(message.to_string()),
+                });
+                // An absent pool.bin is healthy only when another set owns it.
+                env.storage.set_outcome(POOL_META_NAME, None, HealOutcome::FileNotFound);
+                let task_id = ResumeUtils::generate_task_id();
+                if automatic {
+                    ResumeManager::new_replacement_intent(
+                        env.healer.disk.clone(),
+                        task_id.clone(),
+                        "pool_0_set_0".to_string(),
+                        vec![],
+                        vec!["replacement-a".to_string()],
+                        vec![crate::heal::resume::ReplacementTargetIdentity {
+                            endpoint: "replacement-a".to_string(),
+                            canonical_path: "/mnt/replacement-a".to_string(),
+                            physical_device_ids: vec!["device-a".to_string()],
+                            filesystem_identity: "1:2:3".to_string(),
+                        }],
+                    )
+                    .await
+                    .unwrap();
+                }
+                let healer = ErasureSetHealer::new(
+                    env.storage.clone(),
+                    Arc::new(RwLock::new(HealProgress::new())),
+                    CancellationToken::new(),
+                    env.healer.disk.clone(),
+                    HealOpts {
+                        recreate: true,
+                        pool: Some(0),
+                        set: Some(0),
+                        ..Default::default()
+                    },
+                    if automatic {
+                        HealRequestSource::AutoHeal
+                    } else {
+                        HealRequestSource::Admin
+                    },
+                )
+                .with_replacement_targets(vec!["replacement-a".to_string()], automatic.then(|| task_id.clone()));
+                let result = if automatic {
+                    healer.heal_erasure_set(&[], "pool_0_set_0").await
+                } else {
+                    healer
+                        .execute_heal_with_resume(&[], "pool_0_set_0", &env.resume, &env.checkpoint)
+                        .await
+                };
+                let state = if automatic {
+                    ResumeManager::load_replacement_intent(env.healer.disk.clone(), &task_id)
+                        .await
+                        .unwrap()
+                        .get_state()
+                        .await
+                } else {
+                    env.resume.get_state().await
+                };
+                assert_eq!(result.is_ok(), placement == Ok(false));
+                assert_eq!(state.completed, placement == Ok(false));
+                if automatic {
+                    assert_eq!(
+                        state.replacement_phase == crate::heal::resume::ReplacementPhase::Verified,
+                        placement == Ok(false)
+                    );
+                }
+                match placement {
+                    Ok(true) => {
+                        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+                        assert_eq!(state.retry_count, 1);
+                    }
+                    Ok(false) => assert!(env.storage.calls().is_empty()),
+                    Err(message) => {
+                        assert!(result.unwrap_err().to_string().contains(message));
+                        assert!(env.storage.calls().is_empty());
+                    }
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn admin_recreate_target_heals_pool_metadata_before_completion() {
         let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;

crates/heal/src/heal/storage.rs

@@ -436,6 +436,13 @@ pub trait HealStorageAPI: Send + Sync {
         Err(Error::other("target-scoped replacement format is unsupported"))
     }
 
+    /// Whether the explicitly scoped replacement set owns its pool's metadata.
+    /// Backends must use authoritative placement, not treat missing shards as
+    /// evidence that metadata is unnecessary. Unknown placement fails closed.
+    fn replacement_pool_metadata_required(&self, _opts: &HealOpts) -> Result<bool> {
+        Err(Error::other("replacement pool metadata placement is unsupported"))
+    }
+
     /// Read target-specific physical evidence for one replacement version.
     ///
     /// This is only used by automatic replacement healing after the normal
@@ -1268,6 +1275,18 @@ impl HealStorageAPI for ECStoreHealStorage {
             .map_err(Error::Storage)
     }
 
+    fn replacement_pool_metadata_required(&self, opts: &HealOpts) -> Result<bool> {
+        let pool_index = opts
+            .pool
+            .ok_or_else(|| Error::other("replacement pool metadata placement is missing pool scope"))?;
+        let set_index = opts
+            .set
+            .ok_or_else(|| Error::other("replacement pool metadata placement is missing set scope"))?;
+        self.ecstore
+            .replacement_pool_metadata_required(pool_index, set_index)
+            .map_err(Error::Storage)
+    }
+
     async fn replacement_targets_have_version(
         &self,
         bucket: &str,

crates/heal/tests/heal_pool_metadata_scope_test.rs

@@ -0,0 +1,249 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Regression for replacement healing requiring pool.bin on a non-owning set.
+use http::HeaderMap;
+use rustfs_heal::heal::{
+    storage::{ECStoreHealStorage, HealObjectOptions, HealPutObjReader, HealStorageAPI},
+    task::{HealOptions, HealPriority, HealRequest, HealTask, HealType},
+};
+use rustfs_heal_contracts::heal_channel::{HealOpts, HealScanMode};
+use std::{sync::Arc, time::Duration};
+use tokio_util::sync::CancellationToken;
+
+mod storage_api;
+use storage_api::{endpoint_index::*, integration::*, pool_metadata::*};
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn replacement_pool_metadata_follows_real_two_set_placement() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut paths = Vec::new();
+    let mut endpoints = Vec::new();
+    for set in 0..2 {
+        for disk in 0..4 {
+            let path = temp.path().join(format!("set{set}-disk{disk}"));
+            std::fs::create_dir_all(&path).unwrap();
+            let mut endpoint = Endpoint::try_from(path.to_str().unwrap()).unwrap();
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(set);
+            endpoint.set_disk_index(disk);
+            paths.push(path);
+            endpoints.push(endpoint);
+        }
+    }
+    let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints {
+        legacy: false,
+        set_count: 2,
+        drives_per_set: 4,
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "heal-pool-metadata-two-sets".to_string(),
+        platform: "test".to_string(),
+    }]);
+    init_local_disks(endpoint_pools.clone()).await.unwrap();
+    let shutdown = CancellationToken::new();
+    let store = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, shutdown.clone())
+        .await
+        .unwrap();
+    let mut pool_meta = store.pool_meta.read().await.clone();
+    pool_meta.dont_save = false;
+    pool_meta.save(store.pools.clone()).await.unwrap();
+    init_bucket_metadata_sys(store.clone(), vec![]).await;
+    let storage = Arc::new(ECStoreHealStorage::new(store.clone()));
+    let bucket = "metadata-placement";
+    store
+        .make_bucket(
+            bucket,
+            &MakeBucketOptions {
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+
+    for opts in [
+        HealOpts::default(),
+        HealOpts {
+            pool: Some(0),
+            ..Default::default()
+        },
+        HealOpts {
+            set: Some(0),
+            ..Default::default()
+        },
+        HealOpts {
+            pool: Some(1),
+            set: Some(0),
+            ..Default::default()
+        },
+        HealOpts {
+            pool: Some(0),
+            set: Some(2),
+            ..Default::default()
+        },
+    ] {
+        assert!(
+            storage.replacement_pool_metadata_required(&opts).is_err(),
+            "invalid scope must fail closed"
+        );
+    }
+
+    let mut owning_set = None;
+    for set in 0..2 {
+        let opts = HealOpts {
+            recreate: true,
+            scan_mode: HealScanMode::Deep,
+            pool: Some(0),
+            set: Some(set),
+            ..Default::default()
+        };
+        let target_path = &paths[set * 4];
+        let metadata_path = target_path.join(RUSTFS_META_BUCKET).join(POOL_META_NAME);
+        let owns_metadata = metadata_path.join("xl.meta").exists();
+        assert_eq!(storage.replacement_pool_metadata_required(&opts).unwrap(), owns_metadata);
+        if owns_metadata {
+            assert!(owning_set.replace(set).is_none(), "only one set owns pool.bin");
+            std::fs::remove_dir_all(&metadata_path).unwrap();
+        }
+        // Select a user key using the real placement algorithm, independently
+        // of the metadata-scope decision under test.
+        let key = (0..1000)
+            .map(|n| format!("object-{n}"))
+            .find(|key| Arc::ptr_eq(&store.pools[0].get_disks_by_key(key), &store.pools[0].disk_set[set]))
+            .unwrap();
+        let mut versions = Vec::new();
+        for seed in [7, 29] {
+            let bytes: Vec<u8> = (0..(256 * 1024 + 137)).map(|i| ((i + seed) % 251) as u8).collect();
+            let mut reader = HealPutObjReader::from_vec(bytes.clone());
+            let info = store
+                .put_object(
+                    bucket,
+                    &key,
+                    &mut reader,
+                    &HealObjectOptions {
+                        versioned: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .unwrap();
+            versions.push((info.version_id.unwrap().to_string(), bytes));
+        }
+        // Wait for detached PUT owners before deleting their committed shards.
+        {
+            let lock = store.new_ns_lock(bucket, &key).await.unwrap();
+            let _guard = lock.get_write_lock(Duration::from_secs(30)).await.unwrap();
+        }
+        let object_path = target_path.join(bucket).join(&key);
+        let original_parts: Vec<_> = walkdir::WalkDir::new(&object_path)
+            .into_iter()
+            .map(|entry| entry.unwrap())
+            .filter(|entry| entry.file_name().to_str().unwrap().starts_with("part."))
+            .map(|entry| {
+                (
+                    entry.path().strip_prefix(&object_path).unwrap().to_path_buf(),
+                    std::fs::read(entry.path()).unwrap(),
+                )
+            })
+            .collect();
+        assert_eq!(original_parts.len(), 2, "both versions must have physical data shards");
+        std::fs::remove_dir_all(&object_path).unwrap();
+        let target = target_path.to_str().unwrap().to_string();
+        let set_id = format!("pool_0_set_{set}");
+        let mut request = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec![bucket.to_string()],
+                set_disk_id: set_id.clone(),
+            },
+            HealOptions {
+                recreate_missing: true,
+                scan_mode: HealScanMode::Deep,
+                pool_index: Some(0),
+                set_index: Some(set),
+                timeout: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        request.heal_endpoints = vec![target];
+        let task = HealTask::from_request(request, storage.clone());
+        tokio::time::timeout(Duration::from_secs(60), task.execute())
+            .await
+            .unwrap()
+            .expect("both owning and non-owning replacement sets must complete");
+        assert!(
+            !target_path.join(RUSTFS_META_BUCKET).join(HEALING_MARKER_PATH).exists(),
+            "completed repair must clear its healing marker"
+        );
+        assert_eq!(metadata_path.join("xl.meta").exists(), owns_metadata);
+        for (relative, bytes) in original_parts {
+            assert_eq!(std::fs::read(object_path.join(relative)).unwrap(), bytes, "reconstructed shard differs");
+        }
+        for (version, bytes) in versions {
+            let mut reader = store
+                .get_object_reader(
+                    bucket,
+                    &key,
+                    None,
+                    HeaderMap::new(),
+                    &HealObjectOptions {
+                        version_id: Some(version),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .unwrap();
+            let mut actual = Vec::new();
+            tokio::io::copy(&mut reader, &mut actual).await.unwrap();
+            assert_eq!(actual, bytes, "current and historical versions must survive healing");
+        }
+    }
+
+    let set = owning_set.expect("fixture must contain a persisted pool.bin");
+    // Absence in the owning set must never be mistaken for out-of-scope data.
+    for path in &paths[set * 4..set * 4 + 4] {
+        std::fs::remove_dir_all(path.join(RUSTFS_META_BUCKET).join(POOL_META_NAME)).unwrap();
+    }
+    let opts = HealOpts {
+        recreate: true,
+        pool: Some(0),
+        set: Some(set),
+        ..Default::default()
+    };
+    assert!(storage.replacement_pool_metadata_required(&opts).unwrap());
+    let set_id = format!("pool_0_set_{set}");
+    let mut request = HealRequest::new(
+        HealType::ErasureSet {
+            buckets: vec![],
+            set_disk_id: set_id.clone(),
+        },
+        HealOptions {
+            recreate_missing: true,
+            pool_index: Some(0),
+            set_index: Some(set),
+            timeout: Some(Duration::from_secs(60)),
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    request.heal_endpoints = vec![paths[set * 4].to_str().unwrap().to_string()];
+    let task = HealTask::from_request(request, storage);
+    assert!(
+        tokio::time::timeout(Duration::from_secs(60), task.execute())
+            .await
+            .unwrap()
+            .is_err()
+    );
+    shutdown.cancel();
+}

crates/heal/tests/storage_api.rs

@@ -27,3 +27,10 @@ pub(crate) mod integration {
     pub(crate) use rustfs_storage_api::ObjectIO;
     pub(crate) use rustfs_storage_api::ObjectOperations;
 }
+
+#[allow(unused_imports)]
+pub(crate) mod pool_metadata {
+    pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::init_bucket_metadata_sys;
+    pub(crate) use rustfs_ecstore::api::disk::{HEALING_MARKER_PATH, RUSTFS_META_BUCKET};
+    pub(crate) use rustfs_ecstore::api::storage::POOL_META_NAME;
+}