gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-05 19:55:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): preserve automatic replacement recovery status (#7018)

Browse Source 
* 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 <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-09-02 01:21:43 +08:00
committed by GitHub
parent 397dbcf102
commit 7dcfdb3320
15 changed files with 584 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/common.rs
+45
View File
@@ -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<dyn std::error::Error + Send + Sync>> {
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 {
crates/e2e_test/src/replacement_privileged_e2e_test.rs
+189 -138
View File
@@ -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<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync>> {
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<Self, Box<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync>> {
let mut first_error: Option<Box<dyn Error + Send + Sync>> = 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<std::process::Output, Box<dyn Error + Send + Sync>> {
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<PathBuf, Box<dyn Error + Send + Sync>> {
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<u8> {
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::<Vec<_>>();
lines.reverse();
lines.join("\n")
}
fn log_len(path: &Path) -> Result<u64, Box<dyn Error + Send + Sync>> {
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<String, Box<dyn Error + Send + Sync>> {
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<bool, Box<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync>> {
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<bool, Box<dyn Error + Send + Sync>> {
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::<rustfs_filemeta::Error>(),
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<dyn Error + Send + Sync>> {
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<dyn Error + Send + Sync> = Box::new(rustfs_filemeta::Error::FileVersionNotFound);
let missing_file: Box<dyn Error + Send + Sync> = 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()));
crates/ecstore/src/core/pools.rs
+1 -1
View File
@@ -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
crates/ecstore/src/set_disk/mod.rs
+1 -1
View File
@@ -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,
crates/ecstore/src/set_disk/ops/locking.rs
+146 -8
View File
@@ -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<bool> {
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;
crates/heal-contracts/src/heal_channel.rs
+72 -4
View File
@@ -347,6 +347,9 @@ pub struct HealChannelRequest {
pub id: String,
/// Disk ID for heal disk/erasure set task
pub disk: Option<String>,
/// 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<String>,
/// 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<HealChannelPriority>) -> Result<(), String> {
let req = HealChannelRequest {
fn create_auto_heal_disk_request(set_disk_id: String, priority: Option<HealChannelPriority>) -> 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<HealChannelPri
no_lock: None,
timeout_seconds: None,
source: HealRequestSource::AutoHeal,
};
send_heal_request(req).await
}
}
fn create_auto_replacement_disk_request(
pool_index: usize,
set_index: usize,
replacement_endpoint: String,
priority: Option<HealChannelPriority>,
) -> 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<HealChannelPriority>) -> 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<HealChannelPriority>,
) -> 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)]
crates/heal/src/heal/channel.rs
+25 -5
View File
@@ -646,10 +646,12 @@ impl HealChannelProcessor {
/// Convert channel request to heal request
fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result<HealRequest> {
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,
crates/heal/src/heal/mod.rs
+3 -7
View File
@@ -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<L
ecstore_local_disk_map_read().await
}
#[cfg(test)]
pub(crate) type DiskOption = EcstoreDiskOption;
#[cfg(test)]
pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> DiskResult<DiskStore> {
ecstore_new_disk(ep, opt).await
}
crates/heal/src/heal/replacement_readiness.rs
+60 -7
View File
@@ -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<Endpoint> {
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<DiskStore> {
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::<Vec<_>>();
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<Vec<ReplacementTargetIdentity>> {
@@ -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(
crates/heal/src/heal/storage.rs
-9
View File
@@ -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<bool> {
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<bool> {
Ok(super::replacement_readiness::auto_replacement_targets_ready(targets).await)
}
async fn replacement_targets_have_version(
&self,
bucket: &str,
crates/heal/src/heal/storage_api.rs
-2
View File
@@ -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};
}
crates/heal/src/heal/task/heal_erasure_set.rs
-10
View File
@@ -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,
crates/heal/src/heal/task/tests.rs
+8 -12
View File
@@ -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<Option<Error>>,
global_format_calls: Mutex<u32>,
replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
replacement_targets_ready: Mutex<bool>,
replacement_target_identities_ready: Mutex<bool>,
replacement_target_identity_sequences: Mutex<VecDeque<Vec<crate::heal::resume::ReplacementTargetIdentity>>>,
listed_prefixes: Mutex<Vec<String>>,
truncate_without_token: Mutex<bool>,
@@ -943,10 +943,6 @@ impl HealStorageAPI for MockStorage {
))
}
async fn replacement_targets_ready(&self, _targets: &[String]) -> Result<bool> {
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<Vec<crate::heal::resume::ReplacementTargetIdentity>> {
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() {
crates/protos/src/heal_control.rs
+11 -1
View File
@@ -88,6 +88,8 @@ impl From<Priority> for HealChannelPriority {
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct StartCommand {
disk: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
heal_endpoints: Vec<String>,
bucket: String,
object_prefix: Option<String>,
object_version_id: Option<String>,
@@ -113,6 +115,7 @@ impl TryFrom<HealChannelRequest> for StartCommand {
fn try_from(request: HealChannelRequest) -> Result<Self, Self::Error> {
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),
rustfs/src/storage/rpc/node_service.rs
+23
View File
@@ -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));