From 7dcfdb3320b21a7c050a155f0d5b31ab768fd2d8 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 2 Sep 2026 01:21:43 +0800 Subject: [PATCH] fix(heal): preserve automatic replacement recovery status (#7018) * fix(heal): preserve automatic replacement recovery status * fix(heal): admit unformatted replacement targets * fix(heal): preserve replacement heal set scope * fix(heal): attach scoped replacement targets * fix(heal): preserve replacement heal set scope * fix(ecstore): keep startup helper test-only --------- Co-authored-by: houseme --- crates/e2e_test/src/common.rs | 45 +++ .../src/replacement_privileged_e2e_test.rs | 327 ++++++++++-------- crates/ecstore/src/core/pools.rs | 2 +- crates/ecstore/src/set_disk/mod.rs | 2 +- crates/ecstore/src/set_disk/ops/locking.rs | 154 ++++++++- crates/heal-contracts/src/heal_channel.rs | 76 +++- crates/heal/src/heal/channel.rs | 30 +- crates/heal/src/heal/mod.rs | 10 +- crates/heal/src/heal/replacement_readiness.rs | 67 +++- crates/heal/src/heal/storage.rs | 9 - crates/heal/src/heal/storage_api.rs | 2 - crates/heal/src/heal/task/heal_erasure_set.rs | 10 - crates/heal/src/heal/task/tests.rs | 20 +- crates/protos/src/heal_control.rs | 12 +- rustfs/src/storage/rpc/node_service.rs | 23 ++ 15 files changed, 584 insertions(+), 205 deletions(-) diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 5bd11df59..b108cd074 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -1699,6 +1699,51 @@ impl RustFSTestClusterEnvironment { process.wait()?; Ok(()) } + + /// Gracefully stop one cluster node and wait for its process to exit. + /// + /// This is intentionally separate from [`Self::stop_node`]: the latter is + /// a hard kill used by crash-recovery tests, while this path lets RustFS + /// complete its normal shutdown hooks before a test restarts the node. + pub async fn stop_node_gracefully(&mut self, node_idx: usize) -> Result<(), Box> { + self.ensure_node_index(node_idx)?; + + #[cfg(unix)] + { + let Some(process) = self.nodes[node_idx].process.as_ref() else { + return Ok(()); + }; + let pid = process.id().to_string(); + let signal_status = Command::new("kill").args(["-TERM", &pid]).status()?; + if !signal_status.success() { + return Err(format!("failed to send SIGTERM to cluster node {node_idx} (pid {pid})").into()); + } + + let mut process = self.nodes[node_idx] + .process + .take() + .ok_or_else(|| format!("cluster node {node_idx} process disappeared while stopping"))?; + let deadline = std::time::Instant::now() + Duration::from_secs(45); + loop { + if let Some(status) = process.try_wait()? { + info!("Cluster node {} stopped gracefully with {}", node_idx, status); + return Ok(()); + } + if std::time::Instant::now() >= deadline { + let _ = process.kill(); + let _ = process.wait(); + return Err(format!("cluster node {node_idx} did not stop gracefully within 45 seconds").into()); + } + sleep(Duration::from_millis(100)).await; + } + } + + #[cfg(not(unix))] + { + let _ = node_idx; + Err("graceful cluster-node stop is only supported on Unix E2E hosts".into()) + } + } } impl Drop for RustFSTestClusterEnvironment { diff --git a/crates/e2e_test/src/replacement_privileged_e2e_test.rs b/crates/e2e_test/src/replacement_privileged_e2e_test.rs index fc687d4e3..d533e20d0 100644 --- a/crates/e2e_test/src/replacement_privileged_e2e_test.rs +++ b/crates/e2e_test/src/replacement_privileged_e2e_test.rs @@ -40,10 +40,10 @@ mod tests { const ENABLE_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_E2E"; const NAMESPACE_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_E2E_IN_NAMESPACE"; + const LOG_DIR_ENV: &str = "RUSTFS_PRIVILEGED_REPLACEMENT_LOG_DIR"; const TARGET_NODE: usize = 1; const TARGET_DRIVE: usize = 0; const MOUNT_SIZE: &str = "size=128m,mode=0700"; - const ABSENT_SCANNER_OBSERVATION_TIMEOUT_SECS: u64 = 180; const REPLACEMENT_RECOVERY_DIR: &str = ".rustfs.sys/buckets/ahm-replacement"; const REPLACEMENT_INTENT_SUFFIX: &str = "_ahm_replacement_intent.json"; const REPLACEMENT_COMPLETION_PROOF_SUFFIX: &str = "_ahm_replacement_completion_proof.json"; @@ -142,6 +142,23 @@ mod tests { run_command("dmsetup", &["resume", &self.dm_name]) } + fn verify_raw_io_is_unavailable(&self) -> Result<(), Box> { + let mapper = format!("/dev/mapper/{}", self.dm_name); + let output = Command::new("dd") + .env("LC_ALL", "C") + .arg(format!("if={mapper}")) + .args(["of=/dev/null", "bs=4096", "count=1", "iflag=direct", "status=none"]) + .output()?; + if output.status.success() { + return Err(format!("dm-error target unexpectedly allowed a raw read from {mapper}").into()); + } + let stderr = String::from_utf8_lossy(&output.stderr); + if !stderr.contains("Input/output error") { + return Err(format!("raw read from dm-error target failed unexpectedly: {stderr}").into()); + } + Ok(()) + } + fn restore_available(&self) -> Result<(), Box> { let sectors = run_command_stdout("blockdev", &["--getsz", &self.loop_device])?; let linear_table = format!("0 {sectors} linear {} 0", self.loop_device); @@ -194,6 +211,72 @@ mod tests { } } + struct ZramBlockMount { + target: PathBuf, + device: String, + mounted: bool, + } + + impl ZramBlockMount { + fn reserve(target: &Path) -> Result> { + if !Path::new("/dev/zram-control").exists() { + run_command("modprobe", &["zram"])?; + } + let device = run_command_stdout("zramctl", &["--find", "--size", "256M"])?; + if device.is_empty() { + return Err("zramctl --find --size returned an empty device".into()); + } + + Ok(Self { + target: target.to_path_buf(), + device, + mounted: false, + }) + } + + fn mount_target(&mut self) -> Result<(), Box> { + let result = (|| { + run_command("mkfs.ext4", &["-F", &self.device])?; + let target_arg = path_to_string(&self.target, "zram replacement mount target")?; + run_command("mount", &[&self.device, &target_arg]) + })(); + if let Err(error) = result { + let _ = self.cleanup(); + return Err(error); + } + self.mounted = true; + Ok(()) + } + + fn cleanup(&mut self) -> Result<(), Box> { + let mut first_error: Option> = None; + if self.mounted { + if let Err(error) = detach_mount(&self.target) { + first_error.get_or_insert(error); + } else { + self.mounted = false; + } + } + if !self.device.is_empty() { + if let Err(error) = run_command("zramctl", &["--reset", &self.device]) { + first_error.get_or_insert(error); + } else { + self.device.clear(); + } + } + if let Some(error) = first_error { + return Err(error); + } + Ok(()) + } + } + + impl Drop for ZramBlockMount { + fn drop(&mut self) { + let _ = self.cleanup(); + } + } + fn checked_command_output(program: &str, args: &[&str]) -> Result> { let output = Command::new(program).args(args).output()?; if output.status.success() { @@ -298,6 +381,18 @@ mod tests { Err(format!("{ENABLE_ENV}=1 requires root or CAP_SYS_ADMIN; unshare exited with status {status}").into()) } + fn replacement_node_log_path( + cluster_temp_dir: &str, + parity: usize, + node_index: usize, + ) -> Result> { + let log_dir = std::env::var_os(LOG_DIR_ENV) + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from(cluster_temp_dir)); + fs::create_dir_all(&log_dir)?; + Ok(log_dir.join(format!("replacement-ec{parity}-node{node_index}-{}.log", std::process::id()))) + } + fn payload(len: usize, seed: u8) -> Vec { let mut next = seed; (0..len) @@ -472,8 +567,20 @@ mod tests { if let Some(version_id) = &version.version_id { request = request.version_id(version_id); } - let response = request.send().await?; - let body = response.body.collect().await?.into_bytes(); + let response = request.send().await.map_err(|error| { + format!("body GET failed for {}/{}@{:?}: {error}", version.bucket, version.key, version.version_id) + })?; + let body = response + .body + .collect() + .await + .map_err(|error| { + format!( + "body stream failed for {}/{}@{:?}: {error}", + version.bucket, version.key, version.version_id + ) + })? + .into_bytes(); assert_eq!( sha256_hex(&body), *expected_sha256, @@ -582,81 +689,6 @@ mod tests { Ok(()) } - fn log_tail(log: &str) -> String { - let mut lines = log.lines().rev().take(80).collect::>(); - lines.reverse(); - lines.join("\n") - } - - fn log_len(path: &Path) -> Result> { - match fs::metadata(path) { - Ok(metadata) => Ok(metadata.len()), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(0), - Err(error) => Err(format!("failed to stat target node log {path:?}: {error}").into()), - } - } - - fn log_from_offset(path: &Path, offset: u64) -> Result> { - let log = match fs::read(path) { - Ok(log) => log, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(), - Err(error) => return Err(format!("failed to read target node log {path:?}: {error}").into()), - }; - let start = usize::try_from(offset).unwrap_or(usize::MAX).min(log.len()); - Ok(String::from_utf8_lossy(&log[start..]).into_owned()) - } - - fn live_disk_loss_scan_completed(log: &str, target_disk: &Path) -> bool { - let target = target_disk.to_string_lossy(); - let mut saw_live_loss = false; - for line in log.lines() { - if line.contains("Heal auto-scan disk inspection failed") - && line.contains("check_failed") - && line.contains(target.as_ref()) - { - saw_live_loss = true; - continue; - } - if saw_live_loss && (line.contains("Heal auto disk scanner idle") || line.contains("Heal auto-scan cycle completed")) - { - return true; - } - } - false - } - - fn live_disk_loss_scan_completed_from_path( - log_path: &Path, - start_offset: u64, - target_disk: &Path, - ) -> Result> { - Ok(live_disk_loss_scan_completed(&log_from_offset(log_path, start_offset)?, target_disk)) - } - - async fn wait_for_live_disk_loss_observation( - log_path: &Path, - target_disk: &Path, - start_offset: u64, - timeout_secs: u64, - ) -> Result<(), Box> { - let deadline = Instant::now() + Duration::from_secs(timeout_secs); - let mut tick = interval(Duration::from_secs(1)); - loop { - if live_disk_loss_scan_completed_from_path(log_path, start_offset, target_disk)? { - return Ok(()); - } - if Instant::now() >= deadline { - let log = log_from_offset(log_path, start_offset)?; - return Err(format!( - "scanner did not finish a live target-loss scan for {target_disk:?} within {timeout_secs}s; log tail:\n{}", - log_tail(&log) - ) - .into()); - } - tick.tick().await; - } - } - fn cluster_status_is_definitive(status: &serde_json::Value) -> Result> { status["cluster"]["definitive"] .as_bool() @@ -707,6 +739,13 @@ mod tests { .collect() } + fn is_transient_recovery_version_absence(error: &(dyn Error + 'static)) -> bool { + matches!( + error.downcast_ref::(), + Some(rustfs_filemeta::Error::FileVersionNotFound) + ) + } + fn incomplete_versions( target_disk: &Path, versions: &[BaselineVersion], @@ -714,7 +753,21 @@ mod tests { let mut missing = BTreeSet::new(); for version in versions { let actual = - census_object_version_on_disk(target_disk, &version.bucket, &version.key, version.version_id.as_deref())?; + match census_object_version_on_disk(target_disk, &version.bucket, &version.key, version.version_id.as_deref()) { + Ok(actual) => actual, + // During replacement recovery, xl.meta may arrive before this + // particular historical version. The generic census helper + // correctly reports that as an error; this progress poll must + // instead wait for the version to be restored. + Err(error) if is_transient_recovery_version_absence(error.as_ref()) => { + missing.insert(format!( + "{}/{}@{:?}: version metadata not yet present on replacement", + version.bucket, version.key, version.version_id + )); + continue; + } + Err(error) => return Err(error), + }; if !actual.matches_manifest(&version.expected) { missing.insert(format!("{}/{}@{:?}: {actual:?}", version.bucket, version.key, version.version_id)); } @@ -822,13 +875,15 @@ mod tests { let mut mount_ns = MountNamespaceGuard::new()?; let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(3, 4)).await?; - let target_log_path = PathBuf::from(&cluster.temp_dir).join(format!("replacement-node{TARGET_NODE}.log")); - cluster.set_node_capture_log_path(TARGET_NODE, target_log_path.to_string_lossy())?; + for node_index in 0..cluster.nodes.len() { + let node_log_path = replacement_node_log_path(&cluster.temp_dir, parity, node_index)?; + cluster.set_node_capture_log_path(node_index, node_log_path.to_string_lossy())?; + } let target_disk = PathBuf::from(&cluster.nodes[TARGET_NODE].data_dirs[TARGET_DRIVE]); - // Each drive below is an independent tmpfs mount, so this privileged - // path must exercise the production distinct-device/readiness fences. + // The blank target uses a temporary zram block device, so the + // replacement readiness fence sees no root or sibling alias. cluster.extra_env.retain(|(key, _)| key != "RUSTFS_UNSAFE_BYPASS_DISK_CHECK"); - let image_root = PathBuf::from(&cluster.temp_dir).join("replacement-faultable-images"); + let image_root = PathBuf::from(&cluster.temp_dir).join("replacement-block-images"); let mut target_mount = None; for (node_index, node) in cluster.nodes.iter().enumerate() { for (drive_index, drive) in node.data_dirs.iter().enumerate() { @@ -845,6 +900,7 @@ mod tests { } } let mut target_mount = target_mount.ok_or("target drive was not mounted with the faultable block fixture")?; + let mut replacement_mount = ZramBlockMount::reserve(&target_disk)?; cluster.set_env("RUSTFS_HEAL_ENABLED", "true"); cluster.set_env("RUSTFS_SCANNER_ENABLED", "true"); @@ -852,28 +908,34 @@ mod tests { cluster.set_env("RUSTFS_SCANNER_CYCLE", "1"); cluster.set_env("RUSTFS_SCANNER_START_DELAY_SECS", "0"); cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}")); - cluster.set_node_env(TARGET_NODE, "RUST_LOG", "rustfs=info,rustfs::heal::manager=debug,rustfs_notify=debug")?; + for node_index in 0..cluster.nodes.len() { + cluster.set_node_env(node_index, "RUST_LOG", "rustfs=info,rustfs::heal::manager=debug,rustfs_notify=debug")?; + } cluster.start().await?; let clients = cluster.create_all_clients()?; - let versions = seed_baseline(&clients[0], &target_disk).await?; - verify_bodies(&clients[0], &versions).await?; + let versions = seed_baseline(&clients[0], &target_disk) + .await + .map_err(|error| format!("pre-fault baseline seeding failed: {error}"))?; + verify_bodies(&clients[0], &versions) + .await + .map_err(|error| format!("pre-fault body verification failed: {error}"))?; - let live_loss_log_offset = log_len(&target_log_path)?; - target_mount.make_unavailable()?; - wait_for_live_disk_loss_observation( - &target_log_path, - &target_disk, - live_loss_log_offset, - ABSENT_SCANNER_OBSERVATION_TIMEOUT_SECS, - ) - .await?; - assert_no_replacement_status_records(&cluster, &target_disk).await?; - assert_no_replacement_admission_artifacts(&cluster, &target_disk)?; + target_mount + .make_unavailable() + .map_err(|error| format!("failed to install the dm-error target: {error}"))?; + target_mount + .verify_raw_io_is_unavailable() + .map_err(|error| format!("dm-error target was not proven by a direct raw read: {error}"))?; + assert_no_replacement_status_records(&cluster, &target_disk) + .await + .map_err(|error| format!("live-fault replacement status check failed: {error}"))?; + assert_no_replacement_admission_artifacts(&cluster, &target_disk) + .map_err(|error| format!("live-fault replacement artifact check failed: {error}"))?; - cluster.stop_node(TARGET_NODE)?; + cluster.stop_node_gracefully(TARGET_NODE).await?; target_mount.cleanup()?; - mount_ns.mount_tmpfs(&target_disk, &format!("rustfs-e2e-p{parity}-replacement"))?; + replacement_mount.mount_target()?; let missing_before_restart = incomplete_versions(&target_disk, &versions)?; assert_eq!( missing_before_restart.len(), @@ -882,46 +944,26 @@ mod tests { ); cluster.start_node(TARGET_NODE).await?; - wait_for_completed_replacement_with_census(&cluster, &target_disk, &versions, 420).await?; - verify_bodies(&clients[0], &versions).await?; + let recovery_result = async { + wait_for_completed_replacement_with_census(&cluster, &target_disk, &versions, 420).await?; + verify_bodies(&clients[0], &versions).await + } + .await; + let stop_result = cluster.stop_node_gracefully(TARGET_NODE).await; + let replacement_cleanup_result = replacement_mount.cleanup(); - Ok(()) - } + if let Err(error) = recovery_result { + if let Err(stop_error) = stop_result { + info!(%stop_error, "replacement target stop failed while preserving recovery failure"); + } + if let Err(cleanup_error) = replacement_cleanup_result { + info!(%cleanup_error, "replacement zram cleanup failed while preserving recovery failure"); + } + return Err(error); + } + stop_result?; + replacement_cleanup_result?; - #[test] - fn live_loss_barrier_requires_scanner_failure_after_log_offset() -> Result<(), Box> { - let target = Path::new("/mnt/target"); - assert!(live_disk_loss_scan_completed( - "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto-scan cycle completed", - target - )); - assert!(live_disk_loss_scan_completed( - "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle", - target - )); - assert!(!live_disk_loss_scan_completed( - "Heal auto disk scanner idle\nHeal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed", - target - )); - assert!(!live_disk_loss_scan_completed( - "event=disk_health_check_failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle", - target - )); - assert!(!live_disk_loss_scan_completed( - "Heal auto-scan disk inspection failed endpoint=/mnt/other disk_state=check_failed\nHeal auto disk scanner idle", - target - )); - let path = std::env::temp_dir().join(format!("rustfs-replacement-scan-{}.log", std::process::id())); - let stale = - "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n"; - fs::write(&path, stale)?; - let offset = log_len(&path)?; - assert!(!live_disk_loss_scan_completed_from_path(&path, offset, target)?); - let fresh = - "Heal auto-scan disk inspection failed endpoint=/mnt/target disk_state=check_failed\nHeal auto disk scanner idle\n"; - fs::write(&path, format!("{stale}{fresh}"))?; - assert!(live_disk_loss_scan_completed_from_path(&path, offset, target)?); - fs::remove_file(path)?; Ok(()) } @@ -954,6 +996,15 @@ mod tests { ); } + #[test] + fn recovery_census_only_treats_missing_version_as_transient() { + let missing_version: Box = Box::new(rustfs_filemeta::Error::FileVersionNotFound); + let missing_file: Box = Box::new(rustfs_filemeta::Error::FileNotFound); + + assert!(is_transient_recovery_version_absence(missing_version.as_ref())); + assert!(!is_transient_recovery_version_absence(missing_file.as_ref())); + } + #[tokio::test] async fn completion_poll_samples_census_before_status() { let order = std::rc::Rc::new(std::cell::RefCell::new(Vec::new())); diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 483c5633f..46a98ef4b 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -4102,7 +4102,7 @@ impl PoolMetaBootstrapAuthority { } impl PoolMetaWriteState { - #[cfg(any(test, feature = "test-util"))] + #[cfg(test)] pub(crate) fn for_startup(cluster_id: uuid::Uuid, fresh_bootstrap_proven: bool) -> Self { let bootstrap_authority = if fresh_bootstrap_proven { PoolMetaBootstrapAuthority::Fresh diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 34bd6ccb2..8437a8e75 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -127,7 +127,7 @@ use rustfs_filemeta::{ }; use rustfs_heal_contracts::heal_channel::{ DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode, - send_heal_disk, send_heal_request_with_admission, + send_heal_replacement_disk, send_heal_request_with_admission, }; use rustfs_io_metrics::{ record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration, diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index e2674ac38..5a2ea379c 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -22,12 +22,10 @@ use super::super::{ Arc, DiskError, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, Endpoint, Error, FormatV3, HealChannelPriority, LockResult, NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks, StorageError, debug, disk, info, load_format_erasure, - send_heal_disk, warn, + send_heal_replacement_disk, warn, }; -use crate::disk::DiskAPI; use crate::disk::health_state::DriveMembershipSnapshot; -#[cfg(test)] -use crate::disk::new_disk; +use crate::disk::{DiskAPI, new_disk}; use crate::runtime::sources as runtime_sources; use rand::prelude::SliceRandom; #[cfg(test)] @@ -356,11 +354,28 @@ impl SetDisks { Ok(res) => res, Err(e) => { warn!("renew_disk: connect_endpoint err {:?}", &e); - if ep.is_local && e == DiskError::UnformattedDisk { - info!("renew_disk unformatteddisk will trigger heal_disk, {:?}", ep); - let set_disk_id = format!("pool_{}_set_{}", ep.pool_idx, ep.set_idx); - let _ = send_heal_disk(set_disk_id, Some(HealChannelPriority::Normal)).await; + if !matches!(e, DiskError::UnformattedDisk | DiskError::Io(_)) { + return; } + + let attached = match self.attach_unformatted_replacement_disk(ep).await { + Ok(attached) => attached, + Err(err) => { + warn!(endpoint = %ep, error = ?err, "renew_disk: unformatted replacement probe failed"); + return; + } + }; + if !attached { + return; + } + + info!("renew_disk attached unformatted replacement and will trigger heal_disk, {:?}", ep); + let (Ok(pool_index), Ok(set_index)) = (usize::try_from(ep.pool_idx), usize::try_from(ep.set_idx)) else { + warn!("renew_disk: replacement target has invalid pool or set index, {:?}", ep); + return; + }; + let _ = + send_heal_replacement_disk(pool_index, set_index, ep.to_string(), Some(HealChannelPriority::Normal)).await; return; } }; @@ -412,6 +427,60 @@ impl SetDisks { disk_lock[disk_idx] = Some(new_disk); } + /// Attach a replacement target only after proving that the exact local slot + /// is present and unformatted. A health-checked reconnect may reject a + /// blank target before it reaches the format-heal path; that target still + /// has to be visible in this set for the formatter to claim it safely. + async fn attach_unformatted_replacement_disk(&self, ep: &Endpoint) -> disk::error::Result { + if !ep.is_local + || usize::try_from(ep.pool_idx).ok() != Some(self.pool_index) + || usize::try_from(ep.set_idx).ok() != Some(self.set_index) + { + return Ok(false); + } + + let Some(disk_idx) = self.set_endpoints.iter().position(|candidate| candidate == ep) else { + return Ok(false); + }; + + let replacement = new_disk( + ep, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await?; + match load_format_erasure(&replacement, false).await { + Err(DiskError::UnformattedDisk) => {} + Ok(_) => return Ok(false), + Err(err) => return Err(err), + } + + { + let mut disks = self.disks.write().await; + if disks[disk_idx].as_ref().is_some_and(|existing| existing.endpoint() != *ep) { + warn!(endpoint = %ep, disk_idx, "renew_disk rejected unformatted replacement for an occupied foreign slot"); + return Ok(false); + } + disks[disk_idx] = Some(replacement.clone()); + } + + let local_disk_map = runtime_sources::local_disk_map_handle(); + local_disk_map + .write() + .await + .insert(replacement.endpoint().to_string(), Some(replacement.clone())); + + if runtime_sources::setup_is_dist_erasure().await { + let local_disk_set_drives = runtime_sources::local_disk_set_drives_handle(); + let mut local_set_drives = local_disk_set_drives.write().await; + local_set_drives[self.pool_index][self.set_index][disk_idx] = Some(replacement); + } + + Ok(true) + } + pub(in crate::set_disk) fn find_disk_index(&self, fm: &FormatV3) -> Result<(usize, usize)> { self.format.check_other(fm)?; @@ -779,6 +848,75 @@ mod tests { drop(temp_dirs); } + #[tokio::test] + async fn renew_disk_attaches_only_a_verified_local_unformatted_replacement() { + let disk_count = 4; + let format = FormatV3::new(1, disk_count); + let mut temp_dirs = Vec::with_capacity(disk_count); + let mut endpoints = Vec::with_capacity(disk_count); + let mut disks = Vec::with_capacity(disk_count); + + for disk_idx in 0..disk_count - 1 { + let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await; + temp_dirs.push(temp_dir); + endpoints.push(endpoint); + disks.push(Some(disk)); + } + + let replacement_dir = tempfile::tempdir().expect("replacement tempdir should be created"); + let mut replacement_endpoint = + Endpoint::try_from(replacement_dir.path().to_str().expect("replacement path should be utf8")) + .expect("replacement endpoint should parse"); + replacement_endpoint.set_pool_index(0); + replacement_endpoint.set_set_index(0); + replacement_endpoint.set_disk_index(disk_count - 1); + temp_dirs.push(replacement_dir); + endpoints.push(replacement_endpoint.clone()); + disks.push(None); + + let set_disks = SetDisks::new( + "test-owner".to_string(), + Arc::new(RwLock::new(disks)), + disk_count, + disk_count / 2, + 0, + 0, + endpoints, + format, + Vec::new(), + ) + .await; + + assert!( + set_disks + .attach_unformatted_replacement_disk(&replacement_endpoint) + .await + .expect("a blank local replacement should be admitted") + ); + + let attached = set_disks.get_disks_internal().await; + let replacement = attached[disk_count - 1] + .as_ref() + .expect("the verified replacement must occupy its exact set slot"); + assert_eq!(replacement.endpoint(), replacement_endpoint); + assert!( + !replacement.health_check_enabled_for_test(), + "the blank replacement must not start health checks before it receives a format" + ); + assert_eq!( + load_format_erasure(replacement, false).await.unwrap_err(), + DiskError::UnformattedDisk, + "only a still-unformatted replacement may be attached by the fallback" + ); + + runtime_sources::local_disk_map_handle() + .write() + .await + .remove(&replacement_endpoint.to_string()); + drop(set_disks); + drop(temp_dirs); + } + #[tokio::test] async fn renew_disk_rejects_a_format_from_another_slot_or_cluster() { let disk_count = 3; diff --git a/crates/heal-contracts/src/heal_channel.rs b/crates/heal-contracts/src/heal_channel.rs index a5b3327ac..2e4f9a056 100644 --- a/crates/heal-contracts/src/heal_channel.rs +++ b/crates/heal-contracts/src/heal_channel.rs @@ -347,6 +347,9 @@ pub struct HealChannelRequest { pub id: String, /// Disk ID for heal disk/erasure set task pub disk: Option, + /// Exact endpoints of replacement disks for an automatic erasure-set + /// rebuild. An empty list retains the generic erasure-set heal behavior. + pub heal_endpoints: Vec, /// Bucket name pub bucket: String, /// Object prefix (optional) @@ -594,6 +597,7 @@ pub fn create_heal_request( timeout_seconds: None, source: HealRequestSource::Internal, disk: None, + heal_endpoints: Vec::new(), } } @@ -634,12 +638,13 @@ pub fn create_heal_response( } } -pub async fn send_heal_disk(set_disk_id: String, priority: Option) -> Result<(), String> { - let req = HealChannelRequest { +fn create_auto_heal_disk_request(set_disk_id: String, priority: Option) -> HealChannelRequest { + HealChannelRequest { id: Uuid::new_v4().to_string(), bucket: "".to_string(), object_prefix: None, disk: Some(set_disk_id), + heal_endpoints: Vec::new(), object_version_id: None, force_start: false, priority: priority.unwrap_or(HealChannelPriority::Low), @@ -654,8 +659,71 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option, +) -> HealChannelRequest { + let mut request = create_auto_heal_disk_request(format!("pool_{pool_index}_set_{set_index}"), priority); + request.heal_endpoints = vec![replacement_endpoint]; + request.pool_index = Some(pool_index); + request.set_index = Some(set_index); + request +} + +/// Submit the legacy generic erasure-set auto-heal request. +pub async fn send_heal_disk(set_disk_id: String, priority: Option) -> Result<(), String> { + send_heal_request(create_auto_heal_disk_request(set_disk_id, priority)).await +} + +/// Submit an automatic replacement heal for one known disk endpoint. +/// +/// The endpoint makes the request eligible for the durable replacement intent +/// and completion-proof path in the heal task. +pub async fn send_heal_replacement_disk( + pool_index: usize, + set_index: usize, + replacement_endpoint: String, + priority: Option, +) -> Result<(), String> { + send_heal_request(create_auto_replacement_disk_request( + pool_index, + set_index, + replacement_endpoint, + priority, + )) + .await +} + +#[cfg(test)] +mod auto_heal_disk_request_tests { + use super::*; + + #[test] + fn replacement_disk_request_carries_its_exact_endpoint() { + let request = + create_auto_replacement_disk_request(2, 3, "http://node2:9000/drive3".to_string(), Some(HealChannelPriority::Normal)); + + assert_eq!(request.disk.as_deref(), Some("pool_2_set_3")); + assert_eq!(request.heal_endpoints, ["http://node2:9000/drive3"]); + assert_eq!(request.pool_index, Some(2)); + assert_eq!(request.set_index, Some(3)); + assert_eq!(request.source, HealRequestSource::AutoHeal); + } + + #[test] + fn legacy_auto_heal_disk_request_has_no_replacement_endpoint() { + let request = create_auto_heal_disk_request("pool_2_set_3".to_string(), None); + + assert!(request.heal_endpoints.is_empty()); + assert_eq!(request.pool_index, None); + assert_eq!(request.set_index, None); + assert_eq!(request.source, HealRequestSource::AutoHeal); + } } #[cfg(test)] diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index eb03aab9b..33ac33c38 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -646,10 +646,12 @@ impl HealChannelProcessor { /// Convert channel request to heal request fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result { let recursive = request.recursive.unwrap_or(false); + let mut inferred_set_scope = None; let heal_type = if let Some(disk_id) = &request.disk { let set_disk_id = utils::normalize_set_disk_id(disk_id).ok_or_else(|| Error::InvalidHealType { heal_type: format!("erasure-set({disk_id})"), })?; + inferred_set_scope = utils::parse_set_disk_id(&set_disk_id).ok(); HealType::ErasureSet { buckets: vec![], set_disk_id, @@ -712,13 +714,14 @@ impl HealChannelProcessor { dry_run: request.dry_run.unwrap_or(false), no_lock, timeout: request.timeout_seconds.map(std::time::Duration::from_secs), - pool_index: request.pool_index, - set_index: request.set_index, + pool_index: request.pool_index.or_else(|| inferred_set_scope.map(|(pool, _)| pool)), + set_index: request.set_index.or_else(|| inferred_set_scope.map(|(_, set)| set)), }; let mut heal_request = HealRequest::new(heal_type, options, priority); heal_request.id = request.id; heal_request.source = request.source; + heal_request.heal_endpoints = request.heal_endpoints; // 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 @@ -906,6 +909,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, @@ -938,6 +942,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::High, scan_mode: Some(HealScanMode::Normal), remove_corrupted: Some(false), @@ -970,6 +975,7 @@ mod tests { object_prefix: Some("test-object".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::High, scan_mode: Some(HealScanMode::Deep), remove_corrupted: Some(true), @@ -1023,6 +1029,7 @@ mod tests { object_prefix: Some("test-object".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Low, scan_mode: None, remove_corrupted: None, @@ -1060,6 +1067,7 @@ mod tests { object_prefix: Some("test-object".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, @@ -1099,6 +1107,7 @@ mod tests { object_prefix: Some("test-object".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, @@ -1131,6 +1140,7 @@ mod tests { object_prefix: Some("logs/".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::High, scan_mode: Some(HealScanMode::Normal), remove_corrupted: Some(false), @@ -1166,7 +1176,10 @@ mod tests { object_prefix: None, object_version_id: None, disk: Some("pool_0_set_1".to_string()), + heal_endpoints: vec!["http://node0:9000/drive1".to_string()], priority: HealChannelPriority::Critical, + pool_index: Some(0), + set_index: Some(1), scan_mode: None, remove_corrupted: None, recreate_missing: None, @@ -1175,15 +1188,16 @@ mod tests { dry_run: None, no_lock: None, timeout_seconds: None, - pool_index: None, - set_index: None, force_start: false, - source: HealRequestSource::Internal, + source: HealRequestSource::AutoHeal, }; let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); assert!(matches!(heal_request.heal_type, HealType::ErasureSet { .. })); assert_eq!(heal_request.priority, HealPriority::Urgent); + assert_eq!(heal_request.heal_endpoints, ["http://node0:9000/drive1"]); + assert_eq!(heal_request.options.pool_index, Some(0)); + assert_eq!(heal_request.options.set_index, Some(1)); } #[tokio::test] @@ -1197,6 +1211,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: Some("invalid-disk-id".to_string()), + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, @@ -1235,6 +1250,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: channel_priority, scan_mode: None, remove_corrupted: None, @@ -1266,6 +1282,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: Some(false), @@ -1299,6 +1316,7 @@ mod tests { object_prefix: Some("".to_string()), // Empty prefix should be treated as bucket heal object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, @@ -1336,6 +1354,7 @@ mod tests { object_prefix: Some("object".to_string()), object_version_id: None, disk: None, + heal_endpoints: Vec::new(), priority: HealChannelPriority::Low, scan_mode: Some(HealScanMode::Normal), remove_corrupted: None, @@ -1614,6 +1633,7 @@ mod tests { object_prefix: None, object_version_id: None, disk: Some("invalid".to_string()), + heal_endpoints: Vec::new(), priority: HealChannelPriority::Normal, scan_mode: None, remove_corrupted: None, diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 35080e8e9..c918bea49 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -26,12 +26,10 @@ pub mod utils; use storage_api::owner::{ ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET, - EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, - EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, - ecstore_local_disk_map_read, + EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskOption, + EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, + ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk, }; -#[cfg(test)] -use storage_api::owner::{EcstoreDiskOption, ecstore_new_disk}; pub use erasure_healer::ErasureSetHealer; pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts}; @@ -247,10 +245,8 @@ pub(crate) async fn local_disk_map_read() -> tokio::sync::OwnedRwLockReadGuard DiskResult { ecstore_new_disk(ep, opt).await } diff --git a/crates/heal/src/heal/replacement_readiness.rs b/crates/heal/src/heal/replacement_readiness.rs index d238943a0..6cd9d1cc3 100644 --- a/crates/heal/src/heal/replacement_readiness.rs +++ b/crates/heal/src/heal/replacement_readiness.rs @@ -14,9 +14,9 @@ use std::{fs, path::Path}; -#[cfg(test)] -use super::Endpoint; -use super::{DiskStore, HealDiskExt as _, local_disk_map_read, resume::ReplacementTargetIdentity}; +use super::{ + DiskOption, DiskStore, Endpoint, HealDiskExt as _, local_disk_map_read, new_disk, resume::ReplacementTargetIdentity, +}; pub(crate) async fn auto_replacement_target_ready(disk: &DiskStore, local_disks: &[DiskStore]) -> bool { auto_replacement_target_identity(disk, local_disks).await.is_some() @@ -72,8 +72,38 @@ pub(crate) async fn auto_replacement_target_identity( .flatten() } -pub(crate) async fn auto_replacement_targets_ready(targets: &[String]) -> bool { - auto_replacement_target_identities(targets).await.is_some() +fn local_replacement_endpoint(target: &str, local_grid_hosts: &[String]) -> Option { + let mut endpoint = Endpoint::try_from(target).ok()?; + if endpoint.is_local { + return Some(endpoint); + } + + let grid_host = endpoint.grid_host(); + if grid_host.is_empty() || !local_grid_hosts.iter().any(|local_host| local_host == &grid_host) { + return None; + } + + endpoint.is_local = true; + Some(endpoint) +} + +async fn replacement_target_disk(target: &str, local_disks: &[DiskStore]) -> Option { + if let Some(disk) = local_disks.iter().find(|disk| disk.endpoint().to_string() == target) { + return Some(disk.clone()); + } + + let local_grid_hosts = local_disks.iter().map(|disk| disk.endpoint().grid_host()).collect::>(); + let endpoint = local_replacement_endpoint(target, &local_grid_hosts)?; + + new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .ok() } pub(crate) async fn auto_replacement_target_identities(targets: &[String]) -> Option> { @@ -88,8 +118,8 @@ pub(crate) async fn auto_replacement_target_identities(targets: &[String]) -> Op let mut identities = Vec::with_capacity(targets.len()); for target in targets { - let disk = local_disks.iter().find(|disk| disk.endpoint().to_string() == *target)?; - identities.push(auto_replacement_target_identity(disk, &local_disks).await?); + let disk = replacement_target_disk(target, &local_disks).await?; + identities.push(auto_replacement_target_identity(&disk, &local_disks).await?); } identities.sort_by(|left, right| left.endpoint.cmp(&right.endpoint)); identities.dedup_by(|left, right| left.endpoint == right.endpoint); @@ -131,6 +161,29 @@ mod tests { use super::*; use tempfile::TempDir; + #[test] + fn local_replacement_endpoint_accepts_a_url_on_a_registered_local_grid_host() { + let local_grid_hosts = vec!["http://127.0.0.1:9000".to_owned()]; + let endpoint = local_replacement_endpoint("http://127.0.0.1:9000/replacement", &local_grid_hosts) + .expect("matching local grid host should be accepted"); + + assert!(endpoint.is_local); + } + + #[test] + fn local_replacement_endpoint_rejects_a_url_on_an_unregistered_grid_host() { + let local_grid_hosts = vec!["http://127.0.0.1:9000".to_owned()]; + + assert!(local_replacement_endpoint("http://127.0.0.1:9001/replacement", &local_grid_hosts).is_none()); + } + + #[test] + fn local_replacement_endpoint_keeps_a_local_path_local() { + let endpoint = local_replacement_endpoint("/replacement", &[]).expect("local path should be accepted"); + + assert!(endpoint.is_local); + } + #[tokio::test] async fn runtime_environment_cannot_bypass_mount_admission() { temp_env::async_with_vars( diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 6e6e8b06f..b6c5731c9 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -394,11 +394,6 @@ pub trait HealStorageAPI: Send + Sync { Err(Error::other("target-scoped replacement format is unsupported")) } - /// Recheck admitted replacement targets immediately before destructive work. - async fn replacement_targets_ready(&self, _targets: &[String]) -> Result { - Ok(false) - } - /// Read target-specific physical evidence for one replacement version. /// /// This is only used by automatic replacement healing after the normal @@ -1171,10 +1166,6 @@ impl HealStorageAPI for ECStoreHealStorage { .map_err(Error::Storage) } - async fn replacement_targets_ready(&self, targets: &[String]) -> Result { - Ok(super::replacement_readiness::auto_replacement_targets_ready(targets).await) - } - async fn replacement_targets_have_version( &self, bucket: &str, diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index ff722b418..597cb675b 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/crates/heal/src/heal/storage_api.rs @@ -24,7 +24,6 @@ pub(crate) use rustfs_ecstore::api::disk::{ DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH, RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET, }; -#[cfg(test)] pub(crate) use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, new_disk as ecstore_new_disk}; pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageError as EcstoreStorageError}; pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read; @@ -43,7 +42,6 @@ pub(crate) mod owner { EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached, ecstore_local_disk_map_read, }; - #[cfg(test)] pub(crate) use super::{EcstoreDiskOption, ecstore_new_disk}; } diff --git a/crates/heal/src/heal/task/heal_erasure_set.rs b/crates/heal/src/heal/task/heal_erasure_set.rs index df532293a..e9478c721 100644 --- a/crates/heal/src/heal/task/heal_erasure_set.rs +++ b/crates/heal/src/heal/task/heal_erasure_set.rs @@ -93,16 +93,6 @@ impl HealTask { None }; - if is_auto_replacement - && !self - .await_with_control(self.storage.replacement_targets_ready(&self.heal_endpoints)) - .await? - { - return Err(Error::TaskExecutionFailed { - message: format!("Replacement target is no longer ready for automatic heal {set_disk_id}"), - }); - } - let replacement_resume_disk = if is_auto_replacement { Some(match replacement_resume_disk { Some(disk) => disk, diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index c1ed4d6a7..d2846dc5c 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -84,7 +84,7 @@ async fn automatic_replacement_uses_target_scoped_format() { let temp = TempDir::new().expect("temporary resume disk directory should be created"); let disk = make_resume_disk(&temp).await; let storage = Arc::new(MockStorage { - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), resume_disk: Mutex::new(Some(disk)), ..Default::default() }); @@ -123,7 +123,7 @@ async fn automatic_replacement_uses_target_scoped_format() { #[tokio::test] async fn automatic_replacement_persists_intent_before_format() { let storage = Arc::new(MockStorage { - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), ..Default::default() }); let mut request = HealRequest::new( @@ -155,7 +155,7 @@ async fn automatic_replacement_persists_intent_before_format() { #[tokio::test] async fn recovered_replacement_never_uses_a_fresh_resume_disk() { let storage = Arc::new(MockStorage { - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), ..Default::default() }); let mut request = HealRequest::new( @@ -193,7 +193,7 @@ async fn automatic_replacement_rejects_a_new_identity_after_format() { let first_identity = replacement_identity("replacement-a", "device-a", "filesystem-a"); let second_identity = replacement_identity("replacement-a", "device-b", "filesystem-b"); let storage = Arc::new(MockStorage { - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), replacement_target_identity_sequences: Mutex::new(VecDeque::from([ vec![first_identity.clone()], vec![first_identity.clone()], @@ -259,7 +259,7 @@ async fn automatic_replacement_reuses_an_existing_non_target_resume_anchor() { .await .expect("existing intent should be stored on the non-target anchor"); let storage = Arc::new(MockStorage { - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), replacement_resume_disk: Mutex::new(Some(anchor.clone())), ..Default::default() }); @@ -493,7 +493,7 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() { let storage = Arc::new(MockStorage { replacement_resume_disk: Mutex::new(Some(anchor.clone())), - replacement_targets_ready: Mutex::new(true), + replacement_target_identities_ready: Mutex::new(true), ..Default::default() }); let mut request = HealRequest::new( @@ -551,7 +551,7 @@ struct MockStorage { format_error: Mutex>, global_format_calls: Mutex, replacement_format_calls: Mutex)>>, - replacement_targets_ready: Mutex, + replacement_target_identities_ready: Mutex, replacement_target_identity_sequences: Mutex>>, listed_prefixes: Mutex>, truncate_without_token: Mutex, @@ -943,10 +943,6 @@ impl HealStorageAPI for MockStorage { )) } - async fn replacement_targets_ready(&self, _targets: &[String]) -> Result { - Ok(*self.replacement_targets_ready.lock().unwrap()) - } - async fn list_objects_for_heal_page( &self, bucket: &str, @@ -1028,7 +1024,7 @@ impl HealStorageAPI for MockStorage { &self, targets: &[String], ) -> Result> { - if !*self.replacement_targets_ready.lock().unwrap() { + if !*self.replacement_target_identities_ready.lock().unwrap() { return Err(Error::other("replacement target is not ready")); } if let Some(identities) = self.replacement_target_identity_sequences.lock().unwrap().pop_front() { diff --git a/crates/protos/src/heal_control.rs b/crates/protos/src/heal_control.rs index bcb093e90..1742500c5 100644 --- a/crates/protos/src/heal_control.rs +++ b/crates/protos/src/heal_control.rs @@ -88,6 +88,8 @@ impl From for HealChannelPriority { #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct StartCommand { disk: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + heal_endpoints: Vec, bucket: String, object_prefix: Option, object_version_id: Option, @@ -113,6 +115,7 @@ impl TryFrom for StartCommand { fn try_from(request: HealChannelRequest) -> Result { Ok(Self { disk: request.disk, + heal_endpoints: request.heal_endpoints, bucket: request.bucket, object_prefix: request.object_prefix, object_version_id: request.object_version_id, @@ -146,6 +149,7 @@ impl StartCommand { Ok(HealChannelRequest { id: request_id, disk: self.disk, + heal_endpoints: self.heal_endpoints, bucket: self.bucket, object_prefix: self.object_prefix, object_version_id: self.object_version_id, @@ -632,6 +636,12 @@ mod tests { } } + fn replacement_test_request(request_id: String) -> HealChannelRequest { + let mut request = test_request(request_id); + request.heal_endpoints = vec!["http://node1:9000/drive2".to_string()]; + request + } + fn metadata(byte: u8, epoch: u64) -> RequestMetadata { RequestMetadata::new([byte; 16], 1_000, 2_000, epoch) } @@ -639,7 +649,7 @@ mod tests { #[test] fn round_trips_all_commands_and_results() { let request_id = uuid::Uuid::new_v4().to_string(); - let start = Envelope::start(test_request(request_id), metadata(1, 7)).unwrap(); + let start = Envelope::start(replacement_test_request(request_id), metadata(1, 7)).unwrap(); let query = Envelope::query( uuid::Uuid::new_v4().to_string(), metadata(2, 7), diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 71dd5c643..502cec91e 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -230,6 +230,11 @@ fn validate_admin_heal_control_start(request: &rustfs_heal_contracts::heal_chann if request.source != rustfs_heal_contracts::heal_channel::HealRequestSource::Admin { return Err(Status::permission_denied("heal control start source must be admin")); } + if !request.heal_endpoints.is_empty() { + return Err(Status::invalid_argument( + "admin heal control start cannot contain automatic replacement endpoints", + )); + } if request.pool_index.is_some() != request.set_index.is_some() { return Err(Status::invalid_argument("heal control start requires both pool and set")); } @@ -2532,6 +2537,7 @@ mod tests { make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context, make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay, scanner_activity_response_v7, start_decommission_failure_response, stop_rebalance_response, + validate_admin_heal_control_start, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo}; @@ -2766,6 +2772,23 @@ mod tests { assert!(!cache.contains_key("request-1"), "expired idle entries must be purged before admission"); } + #[test] + fn heal_control_admin_start_rejects_automatic_replacement_endpoints() { + let mut request = rustfs_heal_contracts::heal_channel::create_heal_request( + String::new(), + None, + false, + Some(rustfs_heal_contracts::heal_channel::HealChannelPriority::High), + ); + request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Admin; + request.recursive = Some(true); + request.heal_endpoints = vec!["/mnt/replacement".to_string()]; + + let err = validate_admin_heal_control_start(&request) + .expect_err("admin heal-control must not accept automatic replacement targets"); + assert_eq!(err.code(), tonic::Code::InvalidArgument); + } + #[tokio::test] async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() { let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));