From 0875f09a397a7189154ff121cc23ebe2d14eacf9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Tue, 26 May 2026 21:54:29 +0800 Subject: [PATCH] fix: rebuild wiped disks during admin heal (#3084) * fix: rebuild wiped disks during admin heal * Preserve forceStart heal admission semantics * Address heal regression test review comments * fix(heal): address admin heal review comments * fix(heal): fix force-start dedup and test polling * fix(heal): address unresolved review comments * fix(heal): simplify dedup key for clippy --------- Co-authored-by: houseme --- .../src/heal_erasure_disk_rebuild_test.rs | 212 ++++++++++++++++++ crates/e2e_test/src/lib.rs | 3 + crates/ecstore/src/set_disk.rs | 110 +++++---- crates/ecstore/src/set_disk/heal.rs | 18 +- crates/heal/src/heal/channel.rs | 23 +- crates/heal/src/heal/erasure_healer.rs | 10 +- crates/heal/src/heal/manager.rs | 161 +++++++++++-- crates/heal/src/heal/task.rs | 17 +- rustfs/src/admin/handlers/heal.rs | 35 +++ 9 files changed, 499 insertions(+), 90 deletions(-) create mode 100644 crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs new file mode 100644 index 000000000..d4ed96944 --- /dev/null +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -0,0 +1,212 @@ +// 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. + +//! Erasure-set healing regression tests. + +#[cfg(test)] +mod tests { + use crate::common::{RustFSTestEnvironment, execute_awscurl, init_logging}; + use aws_sdk_s3::primitives::ByteStream; + use serial_test::serial; + use std::collections::HashSet; + use std::path::{Path, PathBuf}; + use tokio::time::{Duration, sleep}; + use tracing::info; + + fn has_file_under(path: &Path) -> bool { + let Ok(entries) = std::fs::read_dir(path) else { + return false; + }; + + for entry in entries.filter_map(Result::ok) { + let path = entry.path(); + if path.is_dir() { + if has_file_under(&path) { + return true; + } + } else { + return true; + } + } + + false + } + + fn object_metadata_exists_on_disk(disk: &Path, bucket: &str, key: &str) -> bool { + disk.join(bucket).join(key).join("xl.meta").is_file() + } + + async fn assert_object_body(env: &RustFSTestEnvironment, bucket: &str, key: &str, expected: &[u8]) { + let client = env.create_s3_client(); + let response = client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await + .expect("GET should succeed during/after heal"); + let body = response.body.collect().await.expect("GET body should collect").into_bytes(); + assert_eq!(body.as_ref(), expected, "object body changed for {key}"); + } + + #[tokio::test] + #[serial] + async fn test_admin_deep_heal_rebuilds_cleared_disk_in_single_node_erasure_set() { + init_logging(); + info!("Discussion #2964: admin deep heal should rebuild a wiped disk in a 4-disk single-node erasure set"); + + let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment"); + let root = PathBuf::from(env.temp_dir.clone()); + let disk0 = root.join("disk0"); + let disk1 = root.join("disk1"); + let disk2 = root.join("disk2"); + let disk3 = root.join("disk3"); + for disk in [&disk0, &disk1, &disk2, &disk3] { + std::fs::create_dir_all(disk).expect("disk directory should be created"); + } + + // The test helper always appends env.temp_dir as the final storage path. + // Point it at disk3 and pass the other three disks explicitly. + env.temp_dir = disk3.to_string_lossy().to_string(); + let disk0_arg = disk0.to_string_lossy().to_string(); + let disk1_arg = disk1.to_string_lossy().to_string(); + let disk2_arg = disk2.to_string_lossy().to_string(); + env.start_rustfs_server_with_env( + vec![disk0_arg.as_str(), disk1_arg.as_str(), disk2_arg.as_str()], + &[("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true")], + ) + .await + .expect("Failed to start 4-disk RustFS"); + + let client = env.create_s3_client(); + let bucket = "heal-cleared-disk-regression"; + let target_object_count = std::env::var("RUSTFS_HEAL_REBUILD_OBJECT_COUNT") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(4) + .max(4); + let heal_timeout_secs = std::env::var("RUSTFS_HEAL_REBUILD_TIMEOUT_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(60); + + let mut objects: Vec<(String, Vec, &'static str)> = vec![ + ( + "中文/报告-0001.json".to_string(), + "{\"message\":\"hello 中文\"}".as_bytes().to_vec(), + "application/json", + ), + ( + "english/images/photo-0002.jpg".to_string(), + vec![0xff, 0xd8, 0xff, 0x00, 0x42, 0x24], + "image/jpeg", + ), + ( + "mixed/空 格 + symbols @#%.txt".to_string(), + b"text object with spaces and symbols".to_vec(), + "text/plain; charset=utf-8", + ), + ( + "bin/archive-0004.bin".to_string(), + (0..=255).collect::>(), + "application/octet-stream", + ), + ]; + for index in objects.len()..target_object_count { + objects.push(( + format!("bulk/prefix-{}/object-{index:04}.txt", index % 17), + format!("bulk object {index}: heal regression payload").into_bytes(), + "text/plain; charset=utf-8", + )); + } + + let object_keys = objects.iter().map(|(key, _, _)| key.clone()).collect::>(); + let mut remaining_rebuild_keys: HashSet = object_keys.iter().cloned().collect(); + + client + .create_bucket() + .bucket(bucket) + .send() + .await + .expect("bucket create should succeed"); + for (key, body, content_type) in &objects { + client + .put_object() + .bucket(bucket) + .key(key) + .content_type(*content_type) + .body(ByteStream::from(body.clone())) + .send() + .await + .expect("PUT should succeed"); + } + + assert!(has_file_under(&disk0), "disk0 should contain object shards before wipe"); + env.stop_server(); + + std::fs::remove_dir_all(&disk0).expect("disk0 wipe should succeed"); + std::fs::create_dir_all(&disk0).expect("disk0 should be recreated empty"); + assert!(!has_file_under(&disk0), "disk0 must be empty before restart"); + + env.start_rustfs_server_with_env( + vec![disk0_arg.as_str(), disk1_arg.as_str(), disk2_arg.as_str()], + &[("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true")], + ) + .await + .expect("Failed to restart 4-disk RustFS after disk wipe"); + // The helper's Drop cleanup removes env.temp_dir. Reset it to the parent + // directory after server startup so all four disk directories are cleaned + // without manually deleting a path Drop will also try to remove. + env.temp_dir = root.to_string_lossy().to_string(); + + let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#; + let heal_url = format!("{}/rustfs/admin/v3/heal/{}?forceStart=true", env.url, bucket); + execute_awscurl(&heal_url, "POST", Some(heal_body), &env.access_key, &env.secret_key) + .await + .expect("admin deep heal should be accepted"); + + for _ in 0..heal_timeout_secs { + if !remaining_rebuild_keys.is_empty() { + let mut rebuilt = Vec::new(); + for key in &remaining_rebuild_keys { + if object_metadata_exists_on_disk(&disk0, bucket, key) { + rebuilt.push(key.clone()); + } + } + for key in rebuilt { + let _ = remaining_rebuild_keys.remove(&key); + } + } + + if remaining_rebuild_keys.is_empty() { + for (key, body, _) in &objects { + assert_object_body(&env, bucket, key, body).await; + } + + env.stop_server(); + for key in &object_keys { + assert!( + object_metadata_exists_on_disk(&disk0, bucket, key), + "wiped disk should contain rebuilt xl.meta for {key}" + ); + } + return; + } + + sleep(Duration::from_secs(1)).await; + } + + panic!("admin deep heal did not rebuild all files on the wiped disk within timeout"); + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 39c770260..4eb0acff1 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -114,6 +114,9 @@ mod head_object_range_test; #[cfg(test)] mod head_object_consistency_test; +#[cfg(test)] +mod heal_erasure_disk_rebuild_test; + #[cfg(test)] mod copy_object_metadata_test; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index d1f04f4c0..db6f23065 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -30,7 +30,6 @@ use crate::disk::{ }; use crate::disk::{STORAGE_FORMAT_FILE, count_part_not_success}; use crate::erasure_coding; -use crate::erasure_coding::bitrot_verify; use crate::error::{Error, Result, is_err_version_not_found}; use crate::error::{GenericError, ObjectApiError, is_err_object_not_found}; use crate::global::{GLOBAL_LocalNodeName, GLOBAL_TierConfigMgr}; @@ -4049,32 +4048,16 @@ async fn disks_with_all_parts( continue; } - // Always check data, if we got it. + // Inline data is stored inside xl.meta, so there is no separate part file to + // verify here. Treat the shard as present once metadata was read successfully; + // object reads/heal will validate the inline shard through the normal bitrot + // reader path. Running bitrot_verify directly here can falsely mark small + // inline shards corrupt when older metadata has no per-part checksum entries. if (meta.data.is_some() || meta.size == 0) && !meta.parts.is_empty() { - if let Some(data) = &meta.data { - let checksum_info = meta.erasure.get_checksum_info(meta.parts[0].number); - let checksum_algo = if meta.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { - HashAlgorithm::HighwayHash256SLegacy - } else { - checksum_info.algorithm - }; - let data_len = data.len(); - let verify_err = bitrot_verify( - Box::new(Cursor::new(data.clone())), - data_len, - meta.erasure.shard_file_size(meta.size) as usize, - checksum_algo, - checksum_info.hash, - meta.erasure.shard_size(), - ) - .await - .err(); - - if let Some(vec) = data_errs_by_part.get_mut(&0) - && index < vec.len() - { - vec[index] = conv_part_err_to_int(&verify_err.map(|e| e.into())); - } + if let Some(vec) = data_errs_by_part.get_mut(&0) + && index < vec.len() + { + vec[index] = CHECK_PART_SUCCESS; } continue; } @@ -5393,6 +5376,62 @@ mod tests { assert!(has_part_err(&unknown_errors)); } + #[test] + fn test_populate_data_errs_by_disk_uses_disk_index_not_error_code() { + let mut data_errs_by_disk = HashMap::from([ + (0, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (1, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (2, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + ]); + let data_errs_by_part = HashMap::from([ + (0, vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]), + (1, vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT, CHECK_PART_SUCCESS]), + ]); + + populate_data_errs_by_disk(&mut data_errs_by_disk, &data_errs_by_part); + + assert_eq!(data_errs_by_disk.get(&0).unwrap(), &vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS]); + assert_eq!(data_errs_by_disk.get(&1).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT]); + assert_eq!(data_errs_by_disk.get(&2).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); + + let mut data_errs_by_disk = HashMap::from([ + (0, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (1, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (2, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (3, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + ]); + let data_errs_by_part = HashMap::from([ + ( + 0, + vec![ + CHECK_PART_FILE_NOT_FOUND, + CHECK_PART_SUCCESS, + CHECK_PART_SUCCESS, + CHECK_PART_SUCCESS, + ], + ), + ( + 1, + vec![ + CHECK_PART_FILE_CORRUPT, + CHECK_PART_SUCCESS, + CHECK_PART_SUCCESS, + CHECK_PART_SUCCESS, + ], + ), + ]); + + populate_data_errs_by_disk(&mut data_errs_by_disk, &data_errs_by_part); + + assert_eq!( + data_errs_by_disk.get(&0).unwrap(), + &vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_FILE_CORRUPT] + ); + assert_eq!(data_errs_by_disk.get(&1).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); + assert_eq!(data_errs_by_disk.get(&2).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); + assert_eq!(data_errs_by_disk.get(&3).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); + } + #[test] fn test_should_heal_object_on_disk() { // Test healing decision logic @@ -5413,25 +5452,6 @@ mod tests { assert!(should_heal); } - #[test] - fn test_populate_data_errs_by_disk_uses_disk_index_not_error_code() { - let mut data_errs_by_disk = HashMap::from([ - (0, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), - (1, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), - (2, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), - ]); - let data_errs_by_part = HashMap::from([ - (0, vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]), - (1, vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT, CHECK_PART_SUCCESS]), - ]); - - populate_data_errs_by_disk(&mut data_errs_by_disk, &data_errs_by_part); - - assert_eq!(data_errs_by_disk.get(&0).unwrap(), &vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS]); - assert_eq!(data_errs_by_disk.get(&1).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT]); - assert_eq!(data_errs_by_disk.get(&2).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); - } - #[tokio::test] async fn test_get_disks_info_preserves_runtime_state_for_suspect_and_offline_disks() { let format = FormatV3::new(1, 3); diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index ac5762e6b..9fca89c46 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -350,15 +350,6 @@ impl SetDisks { for (part_index, part) in latest_meta.parts.iter().enumerate() { let till_offset = erasure.shard_file_offset(0, part.size, part.size); - let checksum_info = erasure_info.get_checksum_info(part.number); - let checksum_algo = if latest_meta.uses_legacy_checksum - && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S - { - rustfs_utils::HashAlgorithm::HighwayHash256SLegacy - } else { - checksum_info.algorithm - }; - // Read zero-copy configuration from environment variable // Default: enabled (true) for performance let use_zero_copy = @@ -382,6 +373,15 @@ impl SetDisks { } if let (Some(disk), Some(metadata)) = (disk, ©_parts_metadata[index]) { + let checksum_info = metadata.erasure.get_checksum_info(part.number); + let checksum_algo = if metadata.uses_legacy_checksum + && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S + { + rustfs_utils::HashAlgorithm::HighwayHash256SLegacy + } else { + checksum_info.algorithm + }; + match create_bitrot_reader( metadata.data.as_deref(), Some(disk), diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 0d69120b6..4ad655d60 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -331,7 +331,7 @@ impl HealChannelProcessor { }; // Build HealOptions with all available fields - let mut options = HealOptions { + let options = HealOptions { scan_mode: request.scan_mode.unwrap_or(HealScanMode::Normal), remove_corrupted: request.remove_corrupted.unwrap_or(false), recreate_missing: request.recreate_missing.unwrap_or(true), @@ -343,15 +343,13 @@ impl HealChannelProcessor { set_index: request.set_index, }; - // Apply force_start overrides - if request.force_start { - options.remove_corrupted = true; - options.recreate_missing = true; - options.update_parity = true; - } - let mut heal_request = HealRequest::new(heal_type, options, priority); heal_request.id = request.id; + // force_start controls admission/queue semantics only. Do not reinterpret it as + // destructive heal options: admin clients commonly pass forceStart=true together + // with remove=false, and turning that into remove_corrupted=true can delete the + // remaining healthy bucket volumes before object shards are rebuilt. + heal_request.force_start = request.force_start; Ok(heal_request) } @@ -664,13 +662,14 @@ mod tests { timeout_seconds: None, pool_index: None, set_index: None, - force_start: true, // Should override the above false values + force_start: true, // Admission force only; must not override explicit heal options. }; let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); - assert!(heal_request.options.remove_corrupted); - assert!(heal_request.options.recreate_missing); - assert!(heal_request.options.update_parity); + assert!(heal_request.force_start); + assert!(!heal_request.options.remove_corrupted); + assert!(!heal_request.options.recreate_missing); + assert!(!heal_request.options.update_parity); } #[tokio::test] diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 7ece5cac1..1f46c81f6 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -35,6 +35,7 @@ pub struct ErasureSetHealer { progress: Arc>, cancel_token: tokio_util::sync::CancellationToken, disk: DiskStore, + heal_opts: HealOpts, } impl ErasureSetHealer { @@ -66,12 +67,14 @@ impl ErasureSetHealer { progress: Arc>, cancel_token: tokio_util::sync::CancellationToken, disk: DiskStore, + heal_opts: HealOpts, ) -> Self { Self { storage, progress, cancel_token, disk, + heal_opts, } } @@ -325,6 +328,7 @@ impl ErasureSetHealer { let cancel_token = self.cancel_token.clone(); let in_flight = in_flight.clone(); let set_label = set_disk_id.to_string(); + let heal_opts = self.heal_opts; let permit = semaphore .clone() .acquire_owned() @@ -375,12 +379,6 @@ impl ErasureSetHealer { if !object_exists { Ok(false) } else { - let heal_opts = HealOpts { - scan_mode: HealScanMode::Normal, - remove: true, - recreate: true, - ..Default::default() - }; match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await { Ok((_result, None)) => Ok(true), Ok((_, Some(err))) => Err(Error::other(err)), diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 6c30d0231..439536ecf 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -25,7 +25,7 @@ use rustfs_ecstore::disk::error::DiskError; use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP; use rustfs_madmin::heal_commands::HealResultItem; use std::{ - collections::{BinaryHeap, HashMap, HashSet}, + collections::{BinaryHeap, HashMap}, sync::Arc, time::{Duration, SystemTime}, }; @@ -46,8 +46,8 @@ struct PriorityHealQueue { heap: BinaryHeap, /// Sequence counter for FIFO ordering within same priority sequence: u64, - /// Set of request keys to prevent duplicates - dedup_keys: HashSet, + /// Deduplication key reference counts for queued requests + dedup_keys: HashMap, } /// Wrapper for heap items to implement proper ordering @@ -110,7 +110,7 @@ impl PriorityHealQueue { Self { heap: BinaryHeap::new(), sequence: 0, - dedup_keys: HashSet::new(), + dedup_keys: HashMap::new(), } } @@ -121,7 +121,7 @@ impl PriorityHealQueue { fn pop_next(&mut self) -> Option { self.heap.pop().map(|item| { let key = Self::make_dedup_key(&item.request); - self.dedup_keys.remove(&key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); item.request }) } @@ -133,12 +133,13 @@ impl PriorityHealQueue { fn push(&mut self, request: HealRequest) -> QueuePushOutcome { let key = Self::make_dedup_key(&request); - // Check for duplicates - if self.dedup_keys.contains(&key) { + // Check for duplicates unless the caller explicitly forces admission. + if self.dedup_keys.contains_key(&key) && !request.force_start { return QueuePushOutcome::Merged; } - - self.dedup_keys.insert(key); + // Track dedup keys for both normal and forced requests so queued forced work + // also reserves the dedup key for later non-forced duplicates. + *self.dedup_keys.entry(key).or_insert(0) += 1; self.sequence += 1; self.heap.push(PriorityQueueItem { priority: request.priority, @@ -161,7 +162,7 @@ impl PriorityHealQueue { fn pop(&mut self) -> Option { self.heap.pop().map(|item| { let key = Self::make_dedup_key(&item.request); - self.dedup_keys.remove(&key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); item.request }) } @@ -201,7 +202,7 @@ impl PriorityHealQueue { ( selected.map(|item| { let key = Self::make_dedup_key(&item.request); - self.dedup_keys.remove(&key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); item.request }), skipped, @@ -240,17 +241,27 @@ impl PriorityHealQueue { } } + fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap, key: &str) { + if let Some(count) = dedup_keys.get_mut(key) { + if *count <= 1 { + dedup_keys.remove(key); + } else { + *count -= 1; + } + } + } + /// Check if a request with the same key already exists in the queue #[allow(dead_code)] fn contains_key(&self, request: &HealRequest) -> bool { let key = Self::make_dedup_key(request); - self.dedup_keys.contains(&key) + self.dedup_keys.contains_key(&key) } /// Check if an erasure set heal request for a specific set_disk_id exists fn contains_erasure_set(&self, set_disk_id: &str) -> bool { let key = format!("erasure_set:{set_disk_id}"); - self.dedup_keys.contains(&key) + self.dedup_keys.contains_key(&key) } fn contains_request_id(&self, request_id: &str) -> bool { @@ -277,7 +288,7 @@ impl PriorityHealQueue { while let Some(item) = self.heap.pop() { if removed.is_none() && item.request.id == request_id { let key = Self::make_dedup_key(&item.request); - self.dedup_keys.remove(&key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); removed = Some(item.request); } else { retained.push(item); @@ -298,7 +309,7 @@ impl PriorityHealQueue { while let Some(item) = self.heap.pop() { if should_remove(&item.request) { let key = Self::make_dedup_key(&item.request); - self.dedup_keys.remove(&key); + Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key); removed_count += 1; } else { retained.push(item); @@ -541,7 +552,7 @@ impl HealManager { publish_heal_queue_length(&queue); let queue_capacity = config.queue_size; - if queue.contains_key(&request) { + if !request.force_start && queue.contains_key(&request) { let admission = if request.priority == HealPriority::Low && !config.low_priority_merge_enable { HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) } else { @@ -566,7 +577,7 @@ impl HealManager { return Ok(admission); } - if queue_len >= queue_capacity { + if queue_len >= queue_capacity && !request.force_start { let admission = Self::classify_full_admission(&request, &config); match admission { HealAdmissionResult::Dropped(reason) => { @@ -2124,6 +2135,122 @@ mod tests { ); } + #[tokio::test] + async fn test_force_start_bypasses_duplicate_and_full_admission() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + low_priority_drop_when_full: true, + ..HealConfig::default() + }), + ); + + let normal = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + let mut forced_duplicate = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + forced_duplicate.force_start = true; + + let subsequent_duplicate = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + + assert_eq!( + manager + .submit_heal_request(normal) + .await + .expect("first request should be accepted"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .submit_heal_request(forced_duplicate) + .await + .expect("force start should bypass duplicate/full policy"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .submit_heal_request(subsequent_duplicate) + .await + .expect("subsequent non-force duplicate should be merged"), + HealAdmissionResult::Merged + ); + } + + #[tokio::test] + async fn test_force_start_marks_dedup_key_for_future_duplicates() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + ); + + let normal = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + let mut forced = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + forced.force_start = true; + let duplicate = HealRequest::new( + HealType::Bucket { + bucket: "bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + + assert_eq!( + manager + .submit_heal_request(normal) + .await + .expect("first request should be accepted"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .submit_heal_request(forced) + .await + .expect("forced request should bypass duplicate/full admission"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .submit_heal_request(duplicate) + .await + .expect("non-forced duplicate should merge while forced request is queued"), + HealAdmissionResult::Merged + ); + } + #[test] fn test_running_erasure_set_counts_groups_only_erasure_tasks() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 322025b81..e4a503f0e 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -133,6 +133,8 @@ pub struct HealRequest { pub options: HealOptions, /// Priority pub priority: HealPriority, + /// Whether this request should bypass queue admission dedup/full policies. + pub force_start: bool, /// Created time pub created_at: SystemTime, /// Queue admission time used for scheduler delay metrics @@ -147,6 +149,7 @@ impl HealRequest { heal_type, options, priority, + force_start: false, created_at: now, enqueued_at: now, } @@ -1143,7 +1146,19 @@ impl HealTask { // Step 3: Create erasure set healer with resume support info!("Step 3: Creating erasure set healer with resume support"); - let erasure_healer = ErasureSetHealer::new(self.storage.clone(), self.progress.clone(), self.cancel_token.clone(), disk); + let heal_opts = HealOpts { + recursive: self.options.recursive, + dry_run: self.options.dry_run, + remove: self.options.remove_corrupted, + recreate: self.options.recreate_missing, + scan_mode: self.options.scan_mode, + update_parity: self.options.update_parity, + no_lock: false, + pool: self.options.pool_index, + set: self.options.set_index, + }; + let erasure_healer = + ErasureSetHealer::new(self.storage.clone(), self.progress.clone(), self.cancel_token.clone(), disk, heal_opts); { let mut progress = self.progress.write().await; diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 74e993b91..8108fc365 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -662,6 +662,41 @@ mod tests { assert!(parsed.force_stop); } + #[test] + fn test_heal_channel_request_preserves_admin_heal_options() { + let hip = HealInitParams { + bucket: "bucket".to_string(), + obj_prefix: "prefix".to_string(), + hs: HealOpts { + recursive: true, + dry_run: true, + remove: true, + recreate: true, + scan_mode: HealScanMode::Deep, + update_parity: true, + pool: Some(1), + set: Some(2), + ..Default::default() + }, + force_start: true, + ..Default::default() + }; + + let request = build_heal_channel_request(&hip); + + assert_eq!(request.bucket, "bucket"); + assert_eq!(request.object_prefix.as_deref(), Some("prefix")); + assert!(request.force_start); + assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); + assert_eq!(request.recursive, Some(true)); + assert_eq!(request.dry_run, Some(true)); + assert_eq!(request.remove_corrupted, Some(true)); + assert_eq!(request.recreate_missing, Some(true)); + assert_eq!(request.update_parity, Some(true)); + assert_eq!(request.pool_index, Some(1)); + assert_eq!(request.set_index, Some(2)); + } + #[test] fn test_extract_heal_init_params_allows_root_heal_target() { let uri: Uri = "/rustfs/admin/v3/heal/".parse().expect("uri should parse");