diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 1b4866c74..4a8e52495 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/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 { + 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> { let pools = self.get_pools_for_heal_object(opts)?; diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 9cb5071bc..c172ad6c2 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/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)>>, list_include_lifecycle_object_info: Mutex>, replacement_target_identity_sequences: Mutex>>, + pool_metadata_placement: Mutex>, fail_listing: AtomicBool, fail_listing_buckets: Mutex>, } @@ -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)> { Ok((HealResultItem::default(), None)) } + fn replacement_pool_metadata_required(&self, _opts: &HealOpts) -> Result { + 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; diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 1dc6fdddc..6fb32a5f4 100644 --- a/crates/heal/src/heal/storage.rs +++ b/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 { + 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 { + 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, diff --git a/crates/heal/tests/heal_pool_metadata_scope_test.rs b/crates/heal/tests/heal_pool_metadata_scope_test.rs new file mode 100644 index 000000000..8ff1c505c --- /dev/null +++ b/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 = (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(); +} diff --git a/crates/heal/tests/storage_api.rs b/crates/heal/tests/storage_api.rs index 341834f57..3fb233444 100644 --- a/crates/heal/tests/storage_api.rs +++ b/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; +}