fix(heal): harden replacement rebuild tracking (#5892)

* fix(heal): gate auto replacement formatting

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): require replacement target outcomes

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): bind resumes to replacement targets

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): fence healing marker ownership

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): cover replacement target completion

Co-Authored-By: heihutu <heihutu@gmail.com>

* docs(heal): clarify replacement recovery status

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): canonicalize replacement target checks

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): satisfy marker test module lint

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): scope automatic replacement format

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): require a mounted replacement target

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): avoid cloned ref slice in test

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): revalidate replacement before scanning

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): reset stale resume checkpoints

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): release scanner disk map before probing

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): persist replacement intent before format

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): fail closed on mountinfo read errors

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): fence replacement target identity

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): order replacement completion cleanup

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): atomically seal replacement completion

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): census replacement target shards

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): fence replacement recovery ownership

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): preserve replacement recovery anchors

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): satisfy replacement recovery lint gates

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): bind replacement identity to mount lease

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): cover durable replacement recovery states

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): validate persisted resume task identifiers

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): avoid blocking replacement marker CAS

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): report failed marker rollback

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): pin replacement resume schema compatibility

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): preserve durable recovery anchors

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): preserve public disk path semantics

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): use canonical replacement task ids

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): cover automatic replacement in 3x4 cluster

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): verify replacement target commits

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): persist replacement completion proof

Co-Authored-By: heihutu <heihutu@gmail.com>

* feat(heal): expose durable replacement status

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): bound durable replacement discovery

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): remove replacement readiness bypass

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): retry terminal replacement cleanup

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): isolate replacement intents from legacy resume

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): migrate legacy replacement intents at startup

Co-Authored-By: heihutu <heihutu@gmail.com>

* style(heal): apply strict clippy fix

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): prioritize active replacement recovery state

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): bind readiness to the admitted mount lease

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): atomically publish replacement intents

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): isolate replacement recovery directory

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): tolerate an empty recovery directory

Co-Authored-By: heihutu <heihutu@gmail.com>

* style(heal): remove redundant disk bytes conversion

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): reconcile proof-first replacement recovery

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): fence torn intent recovery

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): cover replacement migration conflicts

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): fence replacement lease mount identity

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(heal): cover missing replacement path admission

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): reject conflicting legacy completion proof

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): fall back to proc mount identity

Co-Authored-By: heihutu <heihutu@gmail.com>

* feat(admin): expose replacement recovery status

Surface the local durable replacement recovery snapshot in the background heal status response so operators can tell whether replacement cleanup is definitive or still pending.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): keep replacement status compatible

Keep the existing background heal status response wire-compatible while retaining the Linux mount lease cleanup needed for the replacement recovery branch.

Co-Authored-By: heihutu <heihutu@gmail.com>

* style(ecstore): match linux mount lease formatting

Keep Linux rustfmt output stable for the replacement mount lease comparison.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): qualify mount lease test constant

Use the disk module path for the format config constant in the Linux mount lease regression test.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): keep procfd mount roots directory-safe

Use a procfd path with an explicit directory component so Unix directory guards can open the replacement mount lease root with O_NOFOLLOW while preserving handle-relative I/O semantics.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): delete empty leased buckets via dirfd

Use the held mount lease fd as the parent for non-force empty bucket deletion on Linux so procfd-rooted paths do not get rejected as BucketNotEmpty. Also make the download-part OpenOptions truncate behavior explicit and keep fsync test recording stable across procfd canonicalization.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): scan leased bucket paths for emptiness

Use the local disk I/O root for bucket emptiness probes before non-force bucket deletion and table-bucket metadata checks. This keeps validation on the same mount instance as the subsequent local disk delete path.

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): align lease path test probes

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): block unsafe replacement recovery restarts

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): defer blocked replacement candidates

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): retry transient replacement discovery

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): keep transient recovery errors retryable

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): block corrupt legacy replacement state

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): classify flat replacement intent corruption

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): keep transient resume loads retryable

Classify malformed legacy replacement state as blocking corruption while preserving disk and transient load failures for retry. This avoids permanently blocking replacement recovery on temporary storage errors.

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): avoid latching transient legacy publishes

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): retry blocked legacy migrations

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(heal): defer blocked startup recoveries

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): preserve disk sync limiter across lease roots

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-08-10 08:32:47 +08:00
committed by GitHub
parent 10a1d6b6e6
commit f17ea7f146
28 changed files with 7060 additions and 442 deletions
+101
View File
@@ -40,6 +40,7 @@ use http::header::{CONTENT_TYPE, HOST};
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::sign_v4;
use s3s::Body;
use std::collections::BTreeSet;
use std::error::Error;
use std::path::{Path, PathBuf};
use tracing::info;
@@ -48,6 +49,34 @@ use walkdir::WalkDir;
type ChaosResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
/// Physical `xl.meta` and shard-file census for one object version on one disk.
///
/// A successful S3 GET only proves that a quorum can serve an object. Replacement
/// tests need this lower-level record to prove that the rebuilt target holds the
/// `xl.meta` selected for a specific version and every `part.N` it declares.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct VersionShardCensus {
pub version_id: Option<String>,
pub has_xl_meta: bool,
pub data_dir: Option<String>,
pub expected_part_numbers: BTreeSet<usize>,
pub present_part_numbers: BTreeSet<usize>,
}
impl VersionShardCensus {
pub(crate) fn is_complete(&self) -> bool {
self.has_xl_meta && self.expected_part_numbers == self.present_part_numbers
}
pub(crate) fn matches_manifest(&self, manifest: &Self) -> bool {
self.version_id == manifest.version_id
&& self.is_complete()
&& manifest.is_complete()
&& self.data_dir == manifest.data_dir
&& self.expected_part_numbers == manifest.expected_part_numbers
}
}
/// Single-node RustFS server with `disk_count` local volume directories that
/// can be faulted individually while the server is running.
pub struct DiskFaultHarness {
@@ -219,6 +248,78 @@ impl DiskFaultHarness {
pub fn object_metadata_exists_on_disk(&self, disk_index: usize, bucket: &str, key: &str) -> bool {
self.disks[disk_index].join(bucket).join(key).join("xl.meta").is_file()
}
/// Census the physical files selected by `version_id` on one disk.
///
/// Missing metadata and missing shard files are represented in the returned
/// census rather than as an error so callers can poll replacement progress.
/// Invalid metadata or an unknown requested version remains an error: treating
/// either as an incomplete rebuild would hide corruption or a wrong-version
/// recovery result.
pub(crate) fn census_object_version(
&self,
disk_index: usize,
bucket: &str,
key: &str,
version_id: Option<&str>,
) -> ChaosResult<VersionShardCensus> {
census_object_version_on_disk(&self.disks[disk_index], bucket, key, version_id)
}
}
/// Census one physical object version without requiring a single-node harness.
/// Cluster replacement tests use the same evidence as the disk-fault tests.
pub(crate) fn census_object_version_on_disk(
disk: &Path,
bucket: &str,
key: &str,
version_id: Option<&str>,
) -> ChaosResult<VersionShardCensus> {
let version_id = version_id.map(str::to_owned);
let object_dir = disk.join(bucket).join(key);
let meta_path = object_dir.join("xl.meta");
if !meta_path.is_file() {
return Ok(VersionShardCensus {
version_id,
has_xl_meta: false,
data_dir: None,
expected_part_numbers: BTreeSet::new(),
present_part_numbers: BTreeSet::new(),
});
}
let metadata = rustfs_filemeta::FileMeta::load(&std::fs::read(&meta_path)?)?;
let file_info = metadata.into_fileinfo(bucket, key, version_id.as_deref().unwrap_or_default(), true, false, true)?;
let expected_part_numbers = if file_info.inline_data() {
BTreeSet::new()
} else {
file_info.parts.iter().map(|part| part.number).collect()
};
let data_dir = file_info.data_dir.map(|id| id.to_string());
let part_dir = data_dir.as_ref().map_or_else(|| object_dir.clone(), |id| object_dir.join(id));
let present_part_numbers = match std::fs::read_dir(&part_dir) {
Ok(entries) => entries
.filter_map(Result::ok)
.filter_map(|entry| {
entry
.file_type()
.ok()
.filter(|kind| kind.is_file())
.and_then(|_| entry.file_name().to_str().map(str::to_owned))
})
.filter_map(|name| name.strip_prefix("part.").and_then(|number| number.parse::<usize>().ok()))
.collect(),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => BTreeSet::new(),
Err(error) => return Err(error.into()),
};
Ok(VersionShardCensus {
version_id,
has_xl_meta: true,
data_dir,
expected_part_numbers,
present_part_numbers,
})
}
/// `POST` a signed (SigV4, service `s3`) admin request without relying on the
@@ -22,15 +22,16 @@
#[cfg(test)]
mod tests {
use crate::chaos::{DiskFaultHarness, signed_admin_post};
use crate::chaos::{DiskFaultHarness, VersionShardCensus, signed_admin_post};
use crate::common::init_logging;
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
use serial_test::serial;
use sha2::{Digest, Sha256};
use std::collections::HashSet;
use tokio::time::{Duration, sleep, timeout};
use std::error::Error;
use tokio::time::{Duration, Instant, interval, timeout};
use tracing::info;
const GET_TIMEOUT: Duration = Duration::from_secs(60);
@@ -271,12 +272,17 @@ mod tests {
put_and_record(&client, bucket, "heal/nested/large.bin", payload(2 * 1024 * 1024, 34), &mut manifest).await?;
verify_manifest(&client, bucket, &manifest, "baseline before disk replacement").await?;
for (key, _) in &manifest {
assert!(
harness.object_metadata_exists_on_disk(0, bucket, key),
"disk0 should hold xl.meta for {key} before replacement"
);
}
let manifest_keys = manifest.iter().map(|(key, _)| key.clone()).collect::<Vec<_>>();
let target_manifest: Vec<(String, VersionShardCensus)> = manifest_keys
.iter()
.map(|key| {
let census = harness.census_object_version(0, bucket, key, None)?;
if !census.is_complete() {
return Err(format!("disk 0 has incomplete physical census for {key}: {census:?}").into());
}
Ok((key.clone(), census))
})
.collect::<Result<_, Box<dyn Error + Send + Sync>>>()?;
harness.kill_server();
harness.replace_disk_with_empty(0)?;
@@ -287,21 +293,112 @@ mod tests {
signed_admin_post(&heal_url, Some(heal_body), &harness.env.access_key, &harness.env.secret_key).await?;
let client = harness.env.create_s3_client();
let mut remaining: HashSet<String> = manifest.iter().map(|(key, _)| key.clone()).collect();
let mut remaining: HashSet<String> = manifest_keys.iter().cloned().collect();
let heal_timeout_secs = std::env::var("RUSTFS_RELIABILITY_HEAL_TIMEOUT_SECS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(120);
let deadline = Instant::now() + Duration::from_secs(heal_timeout_secs);
let mut retry = interval(Duration::from_secs(1));
for _ in 0..heal_timeout_secs {
remaining.retain(|key| !harness.object_metadata_exists_on_disk(0, bucket, key));
loop {
remaining.retain(|key| {
let expected = target_manifest
.iter()
.find(|(manifest_key, _)| manifest_key == key)
.map(|(_, manifest)| manifest)
.expect("every key has a physical manifest");
harness
.census_object_version(0, bucket, key, None)
.map(|census| !census.matches_manifest(expected))
.unwrap_or(true)
});
if remaining.is_empty() {
verify_manifest(&client, bucket, &manifest, "after fresh-disk heal completed").await?;
return Ok(());
}
sleep(Duration::from_secs(1)).await;
if Instant::now() >= deadline {
break;
}
retry.tick().await;
}
Err(format!("fresh-disk heal did not rebuild {remaining:?} on the replaced disk within {heal_timeout_secs}s").into())
}
#[tokio::test]
#[serial]
async fn test_versioned_shard_census_selects_each_version_data_dir() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
info!("Reliability: physical shard census selects the requested object version");
let mut harness = DiskFaultHarness::new(4).await?;
harness.start_server().await?;
let client = harness.env.create_s3_client();
let bucket = "reliability-versioned-census";
let key = "versions/large.bin";
client.create_bucket().bucket(bucket).send().await?;
client
.put_bucket_versioning()
.bucket(bucket)
.versioning_configuration(
VersioningConfiguration::builder()
.status(BucketVersioningStatus::Enabled)
.build(),
)
.send()
.await?;
let first = client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from(payload(256 * 1024, 41)))
.send()
.await?;
let first_version = first.version_id().ok_or("first PUT did not return a version ID")?;
let second = client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from(payload(256 * 1024, 42)))
.send()
.await?;
let second_version = second.version_id().ok_or("second PUT did not return a version ID")?;
let delete = client.delete_object().bucket(bucket).key(key).send().await?;
let delete_version = delete.version_id().ok_or("delete marker did not return a version ID")?;
let first_census = harness.census_object_version(0, bucket, key, Some(first_version))?;
let second_census = harness.census_object_version(0, bucket, key, Some(second_version))?;
let delete_census = harness.census_object_version(0, bucket, key, Some(delete_version))?;
assert!(
first_census.is_complete(),
"first version physical census is incomplete: {first_census:?}"
);
assert!(
second_census.is_complete(),
"second version physical census is incomplete: {second_census:?}"
);
assert_ne!(
first_census.data_dir, second_census.data_dir,
"distinct object versions must select distinct physical data directories"
);
assert_eq!(
first_census.expected_part_numbers, second_census.expected_part_numbers,
"same single-part shape should expose the same part numbers"
);
assert!(
delete_census.is_complete(),
"delete marker physical census is incomplete: {delete_census:?}"
);
assert!(
delete_census.expected_part_numbers.is_empty(),
"delete marker must not declare object shards: {delete_census:?}"
);
assert!(
delete_census.present_part_numbers.is_empty(),
"delete marker must not select stale object shards: {delete_census:?}"
);
Ok(())
}
}
+2 -2
View File
@@ -326,8 +326,8 @@ pub mod disk {
pub use crate::disk::local::ScanGuard;
pub use crate::disk::{
BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize,
CheckPartsResp, ConditionalFileUpdate, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
DiskStore, FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize,
PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
validate_batch_read_version_item_count,
@@ -57,7 +57,7 @@ fn build_part_path(file_path: &Path) -> PathBuf {
async fn open_download_part_file(file_part_path: &Path) -> io::Result<tokio::fs::File> {
let mut options = OpenOptions::new();
options.create(true).read(true).write(true);
options.create(true).truncate(false).read(true).write(true);
#[cfg(not(windows))]
options.mode(0o600);
@@ -784,7 +784,11 @@ impl PeerS3Client for LocalPeerS3Client {
if opts.force_if_empty && !opts.force {
for disk in local_disks.iter() {
if has_xlmeta_files(&disk.path().join(bucket)).await.map_err(Error::Io)? {
let Some(bucket_path) = disk.get_bucket_path_for_io_if_local(bucket) else {
continue;
};
let bucket_path = bucket_path?;
if has_xlmeta_files(&bucket_path).await.map_err(Error::Io)? {
return Err(Error::VolumeNotEmpty);
}
}
+45 -6
View File
@@ -1058,17 +1058,23 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
for (i, set) in new_format_sets.iter().enumerate() {
for (j, fm) in set.iter().enumerate() {
if let Some(fm) = fm {
res.after.drives[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string();
res.after.drives[i * self.set_drive_count + j].state = DriveState::Ok.to_string();
tmp_new_formats[i * self.set_drive_count + j] = Some(fm.clone());
}
}
}
// Save new formats `format.json` on unformatted disks.
for (fm, disk) in tmp_new_formats.iter_mut().zip(disks.iter()) {
if fm.is_some() && disk.is_some() && save_format_file(disk, fm).await.is_err() {
let _ = disk.as_ref().unwrap().close().await;
*fm = None;
for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() {
if fm.is_some() && disk.is_some() {
if let Err(err) = save_format_file(disk, fm).await {
if let Some(disk) = disk.as_ref() {
let _ = disk.close().await;
}
return Ok((res, Some(err.into())));
}
if let Some(saved_format) = fm.as_ref() {
res.after.drives[index].uuid = saved_format.erasure.this.to_string();
res.after.drives[index].state = DriveState::Ok.to_string();
}
}
}
@@ -2206,6 +2212,39 @@ mod tests {
);
}
#[tokio::test]
#[serial]
async fn replacement_format_only_writes_the_requested_slot() {
let (_dirs, _ref_format, sets) = setup_heal_format_sets(1, false).await;
let target = sets.endpoints.endpoints.as_ref()[1].to_string();
let untouched = sets.endpoints.endpoints.as_ref()[2].to_string();
let set = set_level_heal_view(&sets).await;
let (result, error) = set
.heal_replacement_format(false, std::slice::from_ref(&target))
.await
.expect("target-scoped replacement format should run");
assert!(error.is_none(), "target format must not report an error: {error:?}");
assert!(
result
.after
.drives
.iter()
.any(|drive| drive.endpoint == target && drive.state == DriveState::Ok.to_string()),
"requested replacement slot must be formatted"
);
let untouched_format = std::path::Path::new(&sets.endpoints.endpoints.as_ref()[2].get_file_path())
.join(crate::disk::RUSTFS_META_BUCKET)
.join(crate::disk::FORMAT_CONFIG_FILE);
assert!(
!tokio::fs::try_exists(untouched_format)
.await
.expect("untouched replacement format path should be inspectable"),
"unrequested slot {untouched} must remain unformatted"
);
}
fn instance_ctx_test_pool_endpoints() -> (FormatV3, PoolEndpoints) {
let format = FormatV3::new(1, 2);
let endpoints = vec![
+33
View File
@@ -152,6 +152,7 @@ const DISK_OPERATION_NAMES: &[&str] = &[
"read_parts",
"read_multiple",
"write_all",
"compare_and_update_file",
"read_all",
];
@@ -1092,6 +1093,18 @@ impl LocalDiskWrapper {
self.disk.get_object_path(volume, path)
}
pub(crate) fn get_object_path_for_io(&self, volume: &str, path: &str) -> crate::disk::error::Result<std::path::PathBuf> {
self.disk.get_object_path_for_io(volume, path)
}
pub(crate) fn get_bucket_path_for_io(&self, volume: &str) -> crate::disk::error::Result<std::path::PathBuf> {
self.disk.get_bucket_path_for_io(volume)
}
pub fn replacement_mount_lease_root(&self) -> Option<std::path::PathBuf> {
self.disk.replacement_mount_lease_root()
}
pub fn runtime_state(&self) -> RuntimeDriveHealthState {
self.health.runtime_state()
}
@@ -1639,6 +1652,10 @@ impl LocalDiskWrapper {
#[async_trait::async_trait]
impl DiskAPI for LocalDiskWrapper {
fn has_replacement_mount_lease(&self) -> bool {
self.disk.has_replacement_mount_lease()
}
async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
self.track_disk_health_with_op_and_timeout_action(
"read_metadata",
@@ -2140,6 +2157,22 @@ impl DiskAPI for LocalDiskWrapper {
.await
}
async fn compare_and_update_file(
&self,
volume: &str,
path: &str,
expected: Option<Bytes>,
replacement: Option<Bytes>,
) -> Result<crate::disk::ConditionalFileUpdate> {
self.track_disk_health_mutation(
"compare_and_update_file",
DiskMetricMutation::Write,
|| async { self.disk.compare_and_update_file(volume, path, expected, replacement).await },
get_max_timeout_duration(),
)
.await
}
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
self.track_disk_health_with_op(
"read_all",
File diff suppressed because it is too large Load Diff
+89
View File
@@ -115,6 +115,15 @@ pub enum PartTransactionAction {
Rollback,
}
/// Result of an owner-aware file mutation. The disk applies the mutation only
/// while the current contents match the supplied expected value.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ConditionalFileUpdate {
Updated,
Missing,
Mismatch,
}
#[derive(Clone, Copy, Debug)]
pub struct MmapCopyStageMetrics {
pub(crate) path: &'static str,
@@ -557,6 +566,26 @@ impl DiskAPI for Disk {
}
}
async fn compare_and_update_file(
&self,
volume: &str,
path: &str,
expected: Option<Bytes>,
replacement: Option<Bytes>,
) -> Result<ConditionalFileUpdate> {
match self {
Disk::Local(local_disk) => local_disk.compare_and_update_file(volume, path, expected, replacement).await,
Disk::Remote(remote_disk) => remote_disk.compare_and_update_file(volume, path, expected, replacement).await,
}
}
fn has_replacement_mount_lease(&self) -> bool {
match self {
Disk::Local(local_disk) => local_disk.has_replacement_mount_lease(),
Disk::Remote(remote_disk) => remote_disk.has_replacement_mount_lease(),
}
}
#[tracing::instrument(level = "trace", skip_all)]
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
match self {
@@ -695,6 +724,34 @@ impl Disk {
Disk::Remote(_) => None,
}
}
pub(crate) fn get_object_path_for_io_if_local(
&self,
volume: &str,
path: &str,
) -> Option<crate::disk::error::Result<std::path::PathBuf>> {
match self {
Disk::Local(w) => Some(w.get_object_path_for_io(volume, path)),
Disk::Remote(_) => None,
}
}
pub(crate) fn get_bucket_path_for_io_if_local(&self, volume: &str) -> Option<crate::disk::error::Result<std::path::PathBuf>> {
match self {
Disk::Local(w) => Some(w.get_bucket_path_for_io(volume)),
Disk::Remote(_) => None,
}
}
/// Return the descriptor-rooted mount path admitted for automatic
/// replacement, or `None` when the configured endpoint no longer names
/// that held mount instance.
pub fn replacement_mount_lease_root(&self) -> Option<PathBuf> {
match self {
Disk::Local(local_disk) => local_disk.replacement_mount_lease_root(),
Disk::Remote(_) => None,
}
}
}
pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
@@ -860,6 +917,24 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
// CleanAbandonedData
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
/// Atomically replace or remove a small control file only when its current
/// contents match `expected`. Implementations that cannot provide this
/// cross-process guarantee must fail closed instead of emulating it with a
/// read-then-write sequence.
async fn compare_and_update_file(
&self,
_volume: &str,
_path: &str,
_expected: Option<Bytes>,
_replacement: Option<Bytes>,
) -> Result<ConditionalFileUpdate> {
Err(DiskError::MethodNotAllowed)
}
/// Whether local I/O is rooted at a held mount descriptor. Auto-replacement
/// refuses destructive work when this is false.
fn has_replacement_mount_lease(&self) -> bool {
false
}
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo>;
fn start_scan(&self) -> ScanGuard;
}
@@ -1612,6 +1687,7 @@ mod tests {
let endpoint = Endpoint::try_from(test_dir).unwrap();
let local_disk = LocalDisk::new(&endpoint, false).await.unwrap();
let expected_object_path = local_disk.root.join("test-bucket/test-object");
let disk = Disk::Local(Box::new(LocalDiskWrapper::new(Arc::new(local_disk), false)));
// Test basic methods
@@ -1626,6 +1702,19 @@ mod tests {
// Test path method
let path = disk.path();
assert!(path.exists());
let object_path = disk
.get_object_path_if_local("test-bucket", "test-object")
.expect("local disk should expose an object path")
.expect("object path should resolve");
assert_eq!(object_path, expected_object_path);
assert!(!object_path.starts_with("/proc/self/fd/"));
#[cfg(target_os = "linux")]
assert!(
disk.get_object_path_for_io_if_local("test-bucket", "test-object")
.expect("local disk should expose an I/O object path")
.expect("I/O object path should resolve")
.starts_with("/proc/self/fd/")
);
// Test disk location
let location = disk.get_disk_location();
+27 -10
View File
@@ -84,11 +84,22 @@ pub(crate) mod fsync_dir_recorder {
static RECORDED: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
pub(crate) fn record(dir: &Path) {
RECORDED.lock().expect("fsync dir recorder poisoned").push(dir.to_path_buf());
let mut recorded = RECORDED.lock().expect("fsync dir recorder poisoned");
recorded.push(dir.to_path_buf());
if let Ok(canonical) = dir.canonicalize()
&& canonical != dir
{
recorded.push(canonical);
}
}
pub(crate) fn was_fsynced(dir: &Path) -> bool {
RECORDED.lock().expect("fsync dir recorder poisoned").iter().any(|p| p == dir)
let canonical = dir.canonicalize().ok();
RECORDED
.lock()
.expect("fsync dir recorder poisoned")
.iter()
.any(|p| p == dir || canonical.as_ref().is_some_and(|canonical| p == canonical))
}
}
@@ -330,7 +341,7 @@ pub(crate) mod file_sync_probe {
use tokio::sync::Notify;
use tokio::time::timeout;
static ROOT: RwLock<Option<PathBuf>> = RwLock::new(None);
static ROOTS: RwLock<Vec<PathBuf>> = RwLock::new(Vec::new());
static BLOCK_MUTEX: Mutex<()> = Mutex::new(());
static BLOCK_CONDVAR: Condvar = Condvar::new();
static ACTIVE_CHANGED: Notify = Notify::const_new();
@@ -364,7 +375,7 @@ pub(crate) mod file_sync_probe {
fn drop(&mut self) {
release();
FAIL_ON_ATTEMPT.store(usize::MAX, Ordering::SeqCst);
*ROOT.write().expect("file sync probe lock poisoned") = None;
ROOTS.write().expect("file sync probe lock poisoned").clear();
}
}
@@ -377,7 +388,13 @@ pub(crate) mod file_sync_probe {
let _guard = BLOCK_MUTEX.lock().expect("file sync probe blocker poisoned");
BLOCK.store(block, Ordering::SeqCst);
}
*ROOT.write().expect("file sync probe lock poisoned") = Some(root.to_path_buf());
let mut roots = vec![root.to_path_buf()];
if let Ok(canonical) = root.canonicalize()
&& canonical != root
{
roots.push(canonical);
}
*ROOTS.write().expect("file sync probe lock poisoned") = roots;
ProbeGuard
}
@@ -398,11 +415,11 @@ pub(crate) mod file_sync_probe {
}
pub(super) fn enter(path: &Path) -> Option<ActiveGuard> {
let enabled = ROOT
.read()
.expect("file sync probe lock poisoned")
.as_ref()
.is_some_and(|root| path.starts_with(root));
let canonical_path = path.canonicalize().ok();
let enabled =
ROOTS.read().expect("file sync probe lock poisoned").iter().any(|root| {
path.starts_with(root) || canonical_path.as_ref().is_some_and(|canonical| canonical.starts_with(root))
});
if !enabled {
return None;
}
@@ -1439,7 +1439,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
if let Some(disk) = disk_op.as_ref() {
let data_dir = files[idx].data_dir.unwrap_or_default();
let path_str = format!("{object}/{data_dir}/part.{part_number}");
match disk.get_object_path_if_local(bucket, &path_str) {
match disk.get_object_path_for_io_if_local(bucket, &path_str) {
Some(Ok(p)) => batch_items.push((idx, p, adj_off, adj_len)),
_ => return None,
}
+177 -16
View File
@@ -331,6 +331,85 @@ fn warn_heal_writer_failures(
}
impl SetDisks {
/// Read back one healed version from every explicitly admitted replacement
/// target. This is intentionally separate from the normal heal result: a
/// successful result describes the transaction attempt, while automatic
/// replacement completion needs physical evidence that survives a crash
/// before its checkpoint is persisted.
pub(crate) async fn replacement_targets_have_version(
&self,
bucket: &str,
object: &str,
version_id: &str,
targets: &[String],
) -> disk::error::Result<bool> {
let disks = self.get_disks_internal().await;
let mut target_disks = Vec::with_capacity(targets.len());
for target in targets {
let Some(index) = self.set_endpoints.iter().position(|endpoint| endpoint.to_string() == *target) else {
return Ok(false);
};
let Some(disk) = disks.get(index).and_then(Option::as_ref) else {
return Ok(false);
};
target_disks.push(disk.clone());
}
let read_options = ReadOptions {
incl_free_versions: false,
read_data: true,
healing: true,
};
let checks = target_disks.into_iter().map(|disk| {
let read_options = read_options.clone();
async move {
let file_info = match disk.read_version("", bucket, object, version_id, &read_options).await {
Ok(file_info) => file_info,
Err(
DiskError::DiskNotFound
| DiskError::VolumeNotFound
| DiskError::FileNotFound
| DiskError::FileVersionNotFound
| DiskError::PathNotFound,
) => return Ok(false),
Err(err) => return Err(err),
};
if !file_info_is_valid_for_metadata(&file_info) {
return Ok(false);
}
if !version_id.is_empty() && file_info.version_id.as_ref().map(ToString::to_string).as_deref() != Some(version_id)
{
return Ok(false);
}
if file_info.is_canonical_delete_marker() || file_info.is_remote() {
return Ok(true);
}
if (file_info.data.is_some() || file_info.size == 0) && !file_info.parts.is_empty() {
return Ok(true);
}
let check = match disk.check_parts(bucket, object, &file_info).await {
Ok(check) => check,
Err(
DiskError::DiskNotFound
| DiskError::VolumeNotFound
| DiskError::FileNotFound
| DiskError::FileVersionNotFound
| DiskError::PathNotFound,
) => return Ok(false),
Err(err) => return Err(err),
};
Ok(!check.results.is_empty() && check.results.iter().all(|result| *result == CHECK_PART_SUCCESS))
}
});
Ok(futures::future::try_join_all(checks)
.await?
.into_iter()
.all(|committed| committed))
}
#[tracing::instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))]
pub(in crate::set_disk) async fn heal_object(
&self,
@@ -1711,19 +1790,35 @@ impl SetDisks {
}
}
// Heal operation family: the storage-api `HealOperations` contract stays
// implemented `for SetDisks` (contract bounds unchanged) but now lives beside
// its inherent helpers in the `set_disk::ops::heal` module. Bodies are moved
// unchanged; `get_pool_and_set` reads the core through `SetDisksCtx` to keep
// the Heal family aligned with the borrow pattern from #816.
#[async_trait::async_trait]
impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
type Error = Error;
type HealResultItem = HealResultItem;
type HealOptions = HealOpts;
impl SetDisks {
pub(crate) async fn heal_replacement_format(
&self,
dry_run: bool,
targets: &[String],
) -> Result<(HealResultItem, Option<Error>)> {
if targets.is_empty() {
return Err(Error::other("replacement format requires at least one target"));
}
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let mut target_slots = Vec::with_capacity(targets.len());
for target in targets {
let Some(slot) = self.set_endpoints.iter().position(|endpoint| endpoint.to_string() == *target) else {
return Err(Error::other("replacement format target does not belong to the set"));
};
if target_slots.contains(&slot) {
return Err(Error::other("replacement format target is duplicated"));
}
target_slots.push(slot);
}
self.heal_format_for_slots(dry_run, Some(&target_slots)).await
}
async fn heal_format_for_slots(
&self,
dry_run: bool,
target_slots: Option<&[usize]>,
) -> Result<(HealResultItem, Option<Error>)> {
let disks = self.disks.read().await.clone();
let (formats, errs) = load_format_erasure_all(&disks, true).await;
if errs.iter().any(|err| {
@@ -1785,21 +1880,43 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
if !dry_run {
for (disk_idx, err) in errs.iter().enumerate() {
if !matches!(err, Some(DiskError::UnformattedDisk)) {
if !matches!(err, Some(DiskError::UnformattedDisk))
|| target_slots.is_some_and(|slots| !slots.contains(&disk_idx))
{
continue;
}
let mut new_format = ref_format.clone();
new_format.erasure.this = ref_format.erasure.sets[self.set_index][disk_idx];
if save_format_file(&disks[disk_idx], &Some(new_format.clone())).await.is_ok() {
result.after.drives[disk_idx].uuid = new_format.erasure.this.to_string();
result.after.drives[disk_idx].state = DriveState::Ok.to_string();
match save_format_file(&disks[disk_idx], &Some(new_format.clone())).await {
Ok(()) => {
result.after.drives[disk_idx].uuid = new_format.erasure.this.to_string();
result.after.drives[disk_idx].state = DriveState::Ok.to_string();
}
Err(err) => return Ok((result, Some(err.into()))),
}
}
}
Ok((result, None))
}
}
// Heal operation family: the storage-api `HealOperations` contract stays
// implemented `for SetDisks` (contract bounds unchanged) but now lives beside
// its inherent helpers in the `set_disk::ops::heal` module. Bodies are moved
// unchanged; `get_pool_and_set` reads the core through `SetDisksCtx` to keep
// the Heal family aligned with the borrow pattern from #816.
#[async_trait::async_trait]
impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
type Error = Error;
type HealResultItem = HealResultItem;
type HealOptions = HealOpts;
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
self.heal_format_for_slots(dry_run, None).await
}
#[tracing::instrument(skip(self))]
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
@@ -2397,6 +2514,50 @@ mod heal_result_report_tests {
}
}
#[tokio::test]
async fn replacement_target_readback_requires_the_committed_shard() {
let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = "replacement-target-readback";
let object = "object.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let mut reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]);
set.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("source object should be written");
let source = disks[2]
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("source metadata should be readable");
let data_dir = source.data_dir.expect("non-inline source should have a data directory");
let targets = vec![set.set_endpoints[0].to_string(), set.set_endpoints[1].to_string()];
assert!(
set.replacement_targets_have_version(bucket, object, "", &targets)
.await
.expect("healthy target shards should be readable")
);
tokio::fs::remove_file(
temp_dirs[1]
.path()
.join(bucket)
.join(object)
.join(data_dir.to_string())
.join("part.1"),
)
.await
.expect("target shard should be removed after the initial commit");
assert!(
!set.replacement_targets_have_version(bucket, object, "", &targets)
.await
.expect("missing target shard should be observable")
);
}
#[tokio::test]
async fn format_heal_cached_layout_rejects_a_disk_from_another_slot() {
let mut _temp_dirs = Vec::new();
+8 -2
View File
@@ -52,7 +52,10 @@ fn validate_table_bucket_delete_allowed(
async fn table_catalog_metadata_exists(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<bool> {
let local_disks = runtime_sources::local_disks_in(ctx).await;
for disk in local_disks.iter() {
let catalog_path = disk.path().join(bucket).join(BUCKET_TABLE_RESERVED_PREFIX);
let Some(bucket_path) = disk.get_bucket_path_for_io_if_local(bucket) else {
continue;
};
let catalog_path = bucket_path?.join(BUCKET_TABLE_RESERVED_PREFIX);
if has_xlmeta_files(&catalog_path).await? {
return Ok(true);
}
@@ -727,7 +730,10 @@ impl ECStore {
if !opts.force {
let local_disks = runtime_sources::local_disks_in(&self.ctx).await;
for disk in local_disks.iter() {
let bucket_path = disk.path().join(bucket);
let Some(bucket_path) = disk.get_bucket_path_for_io_if_local(bucket) else {
continue;
};
let bucket_path = bucket_path?;
if has_xlmeta_files(&bucket_path).await? {
return Err(StorageError::BucketNotEmpty(bucket.to_string()));
}
+50
View File
@@ -97,6 +97,56 @@ impl ECStore {
Ok((r, None))
}
#[instrument(skip(self, targets), fields(pool_index, set_index, target_count = targets.len()))]
pub async fn heal_replacement_format(
&self,
dry_run: bool,
pool_index: usize,
set_index: usize,
targets: &[String],
) -> Result<(HealResultItem, Option<Error>)> {
let pool = self
.pools
.get(pool_index)
.ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?;
let set = pool.disk_set.get(set_index).cloned().ok_or_else(|| {
StorageError::InvalidArgument(
"heal".to_string(),
"set".to_string(),
format!("invalid heal set index {set_index} for pool {pool_index}"),
)
})?;
set.heal_replacement_format(dry_run, targets).await
}
#[instrument(skip(self, targets), fields(pool_index, set_index, target_count = targets.len()))]
pub async fn replacement_targets_have_version(
&self,
bucket: &str,
object: &str,
version_id: &str,
pool_index: usize,
set_index: usize,
targets: &[String],
) -> Result<bool> {
let pool = self
.pools
.get(pool_index)
.ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?;
let set = pool.disk_set.get(set_index).cloned().ok_or_else(|| {
StorageError::InvalidArgument(
"heal".to_string(),
"set".to_string(),
format!("invalid heal set index {set_index} for pool {pool_index}"),
)
})?;
set.replacement_targets_have_version(bucket, object, version_id, targets)
.await
.map_err(Into::into)
}
#[instrument(skip(self))]
pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let res = self.peer_sys.heal_bucket(bucket, opts).await?;
+559 -27
View File
@@ -14,7 +14,10 @@
use crate::heal::{
progress::HealProgress,
resume::{CheckpointManager, ResumeManager, ResumeUtils, compose_key},
resume::{
CheckpointManager, ReplacementTargetIdentity, ResumeManager, ResumeUtils, compose_key,
replacement_target_identities_match,
},
storage::{HealStorageAPI, next_heal_listing_token},
task::{demote_to_debug_when, is_missing_object_dir_heal_result, take_failure_log_sample},
};
@@ -22,6 +25,7 @@ use crate::{Error, Result};
use futures::{StreamExt, stream::FuturesUnordered};
use metrics::gauge;
use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
use rustfs_madmin::heal_commands::HealResultItem;
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
@@ -85,6 +89,16 @@ pub struct ErasureSetHealer {
disk: DiskStore,
heal_opts: HealOpts,
source: HealRequestSource,
target_endpoints: Arc<[String]>,
replacement_task_id: Option<String>,
replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
}
pub(crate) fn target_outcomes_complete(result: &HealResultItem, target_endpoints: &[String]) -> bool {
target_endpoints.iter().all(|endpoint| {
let mut drives = result.after.drives.iter().filter(|drive| drive.endpoint == *endpoint);
matches!(drives.next(), Some(drive) if drive.state == "ok") && drives.next().is_none()
})
}
impl ErasureSetHealer {
@@ -182,9 +196,46 @@ impl ErasureSetHealer {
disk,
heal_opts,
source,
target_endpoints: Vec::new().into(),
replacement_task_id: None,
replacement_target_identities: None,
}
}
pub(crate) fn with_replacement_targets(
mut self,
mut target_endpoints: Vec<String>,
replacement_task_id: Option<String>,
) -> Self {
target_endpoints.sort_unstable();
target_endpoints.dedup();
self.target_endpoints = target_endpoints.into();
self.replacement_task_id = replacement_task_id;
self
}
pub(crate) fn with_replacement_identity_fence(
mut self,
replacement_target_identities: Option<Vec<ReplacementTargetIdentity>>,
) -> Self {
self.replacement_target_identities = replacement_target_identities.map(Into::into);
self
}
async fn verify_replacement_identity_fence(&self, stage: &str) -> Result<()> {
let Some(expected_identities) = self.replacement_target_identities.as_ref() else {
return Ok(());
};
let actual_identities = self.storage.replacement_target_identities(&self.target_endpoints).await?;
if replacement_target_identities_match(expected_identities, &actual_identities) {
return Ok(());
}
Err(Error::TaskExecutionFailed {
message: format!("Replacement target changed during {stage}"),
})
}
/// execute erasure set heal with resume
#[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
#[hotpath::measure]
@@ -212,9 +263,15 @@ impl ErasureSetHealer {
.await;
result?;
self.verify_replacement_identity_fence("completion").await?;
if self.replacement_task_id.is_some() {
// A replacement marker must outlive the successful data scan. The
// task clears that owner marker before deleting these artifacts.
resume_manager.mark_replacement_completed_and_verified().await?;
return Ok(());
}
// The healing marker is cleared by the caller only after both cleanup
// operations succeed. Cleanup is idempotent, so a retry is safe.
checkpoint_manager.cleanup().await?;
resume_manager.cleanup().await?;
Ok(())
@@ -222,6 +279,21 @@ impl ErasureSetHealer {
/// get or create task id
async fn get_or_create_task_id(&self, set_disk_id: &str) -> Result<String> {
if let Some(task_id) = &self.replacement_task_id {
let manager = ResumeManager::load_replacement_intent(self.disk.clone(), task_id).await?;
let state = manager.get_state().await;
if !state.completed
&& state.set_disk_id == set_disk_id
&& state.replacement_targets.as_slice() == self.target_endpoints.as_ref()
&& state.replacement_generation.as_deref() == Some(task_id.as_str())
{
return Ok(task_id.clone());
}
return Err(Error::TaskExecutionFailed {
message: format!("Replacement resume intent does not match task {task_id}"),
});
}
// check if there are resumable tasks
let resumable_tasks = ResumeUtils::get_resumable_tasks(&self.disk).await?;
@@ -231,6 +303,7 @@ impl ErasureSetHealer {
let state = manager.get_state().await;
if !state.completed
&& state.set_disk_id == set_disk_id
&& state.replacement_targets.as_slice() == self.target_endpoints.as_ref()
&& ResumeUtils::can_resume_task(&self.disk, &task_id).await
{
debug!(
@@ -263,7 +336,7 @@ impl ErasureSetHealer {
}
// create new task id
let task_id = format!("{}_{}", set_disk_id, ResumeUtils::generate_task_id());
let task_id = ResumeUtils::generate_task_id();
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
@@ -285,7 +358,12 @@ impl ErasureSetHealer {
buckets: &[String],
) -> Result<(ResumeManager, CheckpointManager)> {
// check if resume state exists
if ResumeManager::has_resume_state(&self.disk, task_id).await {
let has_resume_state = if self.replacement_task_id.is_some() {
ResumeManager::has_replacement_intent(&self.disk, task_id).await
} else {
ResumeManager::has_resume_state(&self.disk, task_id).await
};
if has_resume_state {
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
@@ -297,7 +375,11 @@ impl ErasureSetHealer {
"Erasure set resume state loading"
);
let resume_manager = ResumeManager::load_from_disk(self.disk.clone(), task_id).await?;
let resume_manager = if self.replacement_task_id.is_some() {
ResumeManager::load_replacement_intent(self.disk.clone(), task_id).await?
} else {
ResumeManager::load_from_disk(self.disk.clone(), task_id).await?
};
let checkpoint_manager = if CheckpointManager::has_checkpoint(&self.disk, task_id).await {
CheckpointManager::load_from_disk(self.disk.clone(), task_id).await?
} else {
@@ -340,6 +422,9 @@ impl ErasureSetHealer {
buckets.to_vec(),
)
.await?;
resume_manager
.set_replacement_targets(self.target_endpoints.as_ref().to_vec())
.await?;
let checkpoint_manager = CheckpointManager::new(self.disk.clone(), task_id.to_string()).await?;
@@ -485,6 +570,12 @@ impl ErasureSetHealer {
// later heal cycle via the same bounded-retry mechanism as failures —
// never hot-retried in place here.
if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 {
if self.replacement_task_id.is_some() && resume_manager.schedule_retry().await? {
checkpoint_manager.reset_for_retry().await?;
return Err(Error::transient_skip(format!(
"Replacement erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
)));
}
if resume_manager.schedule_retry().await? {
// Both persistence layers must be reset together: schedule_retry
// rewinds the resume state (cursor + counters), and the
@@ -508,15 +599,15 @@ impl ErasureSetHealer {
state = "retry_scheduled",
"Erasure set heal pass finished with unhealed versions; scheduled full re-heal retry"
);
return Err(Error::other(format!(
return Err(Error::transient_skip(format!(
"Erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
)));
}
// Retry budget exhausted: drop the resume/checkpoint state so this
// task does not loop, but keep the healing markers (return Err) so a
// later heal cycle / the background scanner starts a fresh attempt.
// Never silently claim a clean completion while objects are unhealed.
// Retry budget exhausted: keep the resume/checkpoint state while
// the replacement marker remains. A later repair must retain the
// durable evidence of the incomplete generation instead of
// starting from an indistinguishable blank state.
error!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
@@ -529,15 +620,16 @@ impl ErasureSetHealer {
state = "failed_after_retries",
"Erasure set heal exhausted retries with unrecovered versions"
);
checkpoint_manager.cleanup().await?;
resume_manager.cleanup().await?;
return Err(Error::other(format!(
"Erasure set heal exhausted retries with {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped"
)));
}
// no failures — mark task completed
resume_manager.mark_completed().await?;
// No failures — ordinary heals are complete now. Replacement heals
// atomically transition to Verified after the terminal identity fence.
if self.replacement_task_id.is_none() {
resume_manager.mark_completed().await?;
}
debug!(
target: "rustfs::heal::erasure_healer",
@@ -628,6 +720,7 @@ impl ErasureSetHealer {
matches!(self.heal_opts.scan_mode, HealScanMode::Deep) || matches!(self.source, HealRequestSource::AutoHeal);
loop {
self.verify_replacement_identity_fence("page scan").await?;
// Get one page of object versions
let (objects, next_token, is_truncated) = if use_disk_walk {
self.storage
@@ -672,6 +765,8 @@ impl ErasureSetHealer {
let set_label = set_disk_id.to_string();
let heal_opts = self.heal_opts;
let semaphore = semaphore.clone();
let target_endpoints = self.target_endpoints.clone();
let replacement_commit_evidence_required = self.replacement_task_id.is_some();
page_tasks.push(async move {
let permit = semaphore
@@ -699,6 +794,35 @@ impl ErasureSetHealer {
.heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts)
.await
{
Ok((result, None))
if target_outcomes_complete(&result, &target_endpoints) =>
{
if !replacement_commit_evidence_required {
Ok(true)
} else {
match storage
.replacement_targets_have_version(
&bucket_name,
&object_name,
version_id.as_deref(),
&heal_opts,
&target_endpoints,
)
.await
{
Ok(true) => Ok(true),
Ok(false) => Err(Error::transient_skip(format!(
"Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version"
))),
Err(err) => Err(Error::transient_skip(format!(
"Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}"
))),
}
}
}
Ok((_result, None)) if !target_endpoints.is_empty() => Err(Error::transient_skip(format!(
"Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed"
))),
Ok((_result, None)) => Ok(true),
Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false),
Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) {
@@ -1011,9 +1135,12 @@ mod resume_loop_tests {
//! that emits programmable multi-version pages. These exercise the real loop
//! logic (cursor seeding, per-version dedup, anti-loop guard, absence
//! handling) — not merely a mock's own output.
use super::ErasureSetHealer;
use super::{ErasureSetHealer, target_outcomes_complete};
use crate::heal::progress::HealProgress;
use crate::heal::resume::{CheckpointManager, RESUME_CHECKPOINT_FILE, ResumeDeleteFailure, ResumeManager, compose_key};
use crate::heal::resume::{
CheckpointManager, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure, ResumeManager, ResumeUtils,
compose_key,
};
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage_api::status::BucketInfo;
use crate::heal::{
@@ -1021,8 +1148,8 @@ mod resume_loop_tests {
};
use crate::{Error, Result};
use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
use rustfs_madmin::heal_commands::HealResultItem;
use std::collections::HashMap;
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos};
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tempfile::TempDir;
@@ -1037,6 +1164,53 @@ mod resume_loop_tests {
}
}
#[test]
fn target_outcomes_require_each_requested_endpoint_once_and_ok() {
let result = HealResultItem {
after: Infos {
drives: vec![
HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "ok".to_string(),
..Default::default()
},
HealDriveInfo {
endpoint: "replacement-b".to_string(),
state: "missing".to_string(),
..Default::default()
},
],
},
..Default::default()
};
assert!(target_outcomes_complete(&result, &["replacement-a".to_string()]));
assert!(!target_outcomes_complete(
&result,
&["replacement-a".to_string(), "replacement-b".to_string()]
));
assert!(!target_outcomes_complete(&result, &["replacement-c".to_string()]));
let duplicate = HealResultItem {
after: Infos {
drives: vec![
HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "ok".to_string(),
..Default::default()
},
HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "missing".to_string(),
..Default::default()
},
],
},
..Default::default()
};
assert!(!target_outcomes_complete(&duplicate, &["replacement-a".to_string()]));
}
#[derive(Clone)]
struct Page {
items: Vec<HealListItem>,
@@ -1061,8 +1235,14 @@ mod resume_loop_tests {
pages: Mutex<HashMap<Option<String>, Page>>,
/// per-`compose_key` heal outcome; default is `Ok`
outcomes: Mutex<HashMap<String, HealOutcome>>,
/// successful low-level result per `compose_key`; default has no drive outcomes.
results: Mutex<HashMap<String, HealResultItem>>,
/// Target-specific physical readback evidence per `compose_key`; the
/// fake models a healthy backend unless a test explicitly revokes it.
replacement_commit_evidence: Mutex<HashMap<String, bool>>,
/// every heal_object call recorded as (name, version_id)
heal_calls: Mutex<Vec<(String, Option<String>)>>,
replacement_target_identity_sequences: Mutex<VecDeque<Vec<ReplacementTargetIdentity>>>,
fail_listing: AtomicBool,
}
@@ -1073,6 +1253,15 @@ mod resume_loop_tests {
fn set_outcome(&self, name: &str, version: Option<&str>, outcome: HealOutcome) {
self.outcomes.lock().unwrap().insert(compose_key(name, version), outcome);
}
fn set_result(&self, name: &str, version: Option<&str>, result: HealResultItem) {
self.results.lock().unwrap().insert(compose_key(name, version), result);
}
fn set_replacement_commit_evidence(&self, name: &str, version: Option<&str>, committed: bool) {
self.replacement_commit_evidence
.lock()
.unwrap()
.insert(compose_key(name, version), committed);
}
fn calls(&self) -> Vec<(String, Option<String>)> {
self.heal_calls.lock().unwrap().clone()
}
@@ -1143,7 +1332,7 @@ mod resume_loop_tests {
let key = compose_key(object, version_id);
let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
match outcome {
HealOutcome::Ok => Ok((HealResultItem::default(), None)),
HealOutcome::Ok => Ok((self.results.lock().unwrap().get(&key).cloned().unwrap_or_default(), None)),
HealOutcome::VersionNotFound => {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
}
@@ -1157,6 +1346,21 @@ mod resume_loop_tests {
async fn heal_format(&self, _dry: bool) -> Result<(HealResultItem, Option<Error>)> {
Ok((HealResultItem::default(), None))
}
async fn replacement_targets_have_version(
&self,
_bucket: &str,
object: &str,
version_id: Option<&str>,
_opts: &HealOpts,
_targets: &[String],
) -> Result<bool> {
Ok(*self
.replacement_commit_evidence
.lock()
.unwrap()
.get(&compose_key(object, version_id))
.unwrap_or(&true))
}
async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result<Vec<HealListItem>> {
Ok(Vec::new())
}
@@ -1179,6 +1383,13 @@ mod resume_loop_tests {
async fn get_disk_for_resume(&self, _id: &str) -> Result<DiskStore> {
Err(Error::other("not implemented in tests"))
}
async fn replacement_target_identities(&self, _targets: &[String]) -> Result<Vec<ReplacementTargetIdentity>> {
self.replacement_target_identity_sequences
.lock()
.unwrap()
.pop_front()
.ok_or_else(|| Error::other("replacement identity sequence exhausted"))
}
}
async fn make_disk(temp: &TempDir) -> DiskStore {
@@ -1204,13 +1415,19 @@ mod resume_loop_tests {
storage: Arc<FakeStorage>,
resume: ResumeManager,
checkpoint: CheckpointManager,
task_id: String,
_temp: TempDir,
}
async fn make_env() -> Env {
make_env_with_targets(Vec::new()).await
}
async fn make_env_with_targets(target_endpoints: Vec<String>) -> Env {
let temp = TempDir::new().unwrap();
let disk = make_disk(&temp).await;
let storage = Arc::new(FakeStorage::default());
let task_id = ResumeUtils::generate_task_id();
let healer = ErasureSetHealer::new(
storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
@@ -1218,22 +1435,24 @@ mod resume_loop_tests {
disk.clone(),
HealOpts::default(),
HealRequestSource::Internal,
);
)
.with_replacement_targets(target_endpoints, None);
let resume = ResumeManager::new(
disk.clone(),
"task".to_string(),
task_id.clone(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
vec!["b".to_string()],
)
.await
.unwrap();
let checkpoint = CheckpointManager::new(disk, "task".to_string()).await.unwrap();
let checkpoint = CheckpointManager::new(disk, task_id.clone()).await.unwrap();
Env {
healer,
storage,
resume,
checkpoint,
task_id,
_temp: temp,
}
}
@@ -1277,6 +1496,112 @@ mod resume_loop_tests {
assert_eq!(env.resume.resume_cursor().await, None);
}
#[tokio::test]
async fn replacement_targets_use_a_canonical_order() {
let env = make_env_with_targets(vec![
"replacement-b".to_string(),
"replacement-a".to_string(),
"replacement-b".to_string(),
])
.await;
assert_eq!(env.healer.target_endpoints.as_ref(), ["replacement-a", "replacement-b"]);
}
#[tokio::test]
async fn replacement_identity_fence_rejects_a_remount_before_page_scan() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let expected_identity = ReplacementTargetIdentity {
endpoint: "replacement-a".to_string(),
canonical_path: "/mnt/replacement-a".to_string(),
physical_device_ids: vec!["device-a".to_string()],
filesystem_identity: "filesystem-a".to_string(),
};
let remounted_identity = ReplacementTargetIdentity {
physical_device_ids: vec!["device-b".to_string()],
filesystem_identity: "filesystem-b".to_string(),
..expected_identity.clone()
};
env.storage
.replacement_target_identity_sequences
.lock()
.unwrap()
.push_back(vec![remounted_identity]);
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts::default(),
HealRequestSource::AutoHeal,
)
.with_replacement_targets(vec!["replacement-a".to_string()], Some("generation-a".to_string()))
.with_replacement_identity_fence(Some(vec![expected_identity]));
let mut current_object_index = 0;
let mut processed = 0;
let mut successful = 0;
let mut failed = 0;
let mut skipped = 0;
let error = healer
.heal_bucket_with_resume(
"b",
"pool_0_set_0",
0,
&mut current_object_index,
&mut processed,
&mut successful,
&mut failed,
&mut skipped,
&env.resume,
&env.checkpoint,
)
.await
.expect_err("a remounted target must not begin a new page scan");
assert!(error.to_string().contains("page scan"));
assert!(env.storage.calls().is_empty());
}
#[tokio::test]
async fn replacement_generation_never_reuses_another_disk_cursor() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
ResumeManager::new_replacement_intent(
env.healer.disk.clone(),
ResumeUtils::generate_task_id(),
"pool_0_set_0".to_string(),
vec!["b".to_string()],
vec!["replacement-a".to_string()],
vec![crate::heal::resume::ReplacementTargetIdentity {
endpoint: "replacement-a".to_string(),
canonical_path: "/mnt/replacement-a".to_string(),
physical_device_ids: vec!["device-a".to_string()],
filesystem_identity: "1:2:3".to_string(),
}],
)
.await
.expect("first replacement intent should persist");
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts::default(),
HealRequestSource::AutoHeal,
)
.with_replacement_targets(vec!["replacement-a".to_string()], Some(ResumeUtils::generate_task_id()));
let error = healer
.get_or_create_task_id("pool_0_set_0")
.await
.expect_err("a second replacement must not reuse the first replacement cursor");
assert!(
!error.to_string().contains("generation-a"),
"the previous replacement generation must not be selected"
);
}
#[tokio::test]
async fn object_timeout_aborts_the_bucket_page_immediately() {
let env = make_env().await;
@@ -1330,13 +1655,14 @@ mod resume_loop_tests {
.await
.expect("new heal should allocate a task id");
assert_ne!(task_id, "task", "a completed resume state must not suppress a new heal");
assert_ne!(task_id, env.task_id, "a completed resume state must not suppress a new heal");
assert!(uuid::Uuid::parse_str(&task_id).is_ok(), "new resume task ids must be UUIDs");
}
#[tokio::test]
async fn cleanup_failure_keeps_erasure_set_heal_incomplete() {
let env = make_env().await;
let checkpoint_path = format!("{BUCKET_META_PREFIX}/task_{RESUME_CHECKPOINT_FILE}");
let checkpoint_path = format!("{BUCKET_META_PREFIX}/{}_{RESUME_CHECKPOINT_FILE}", env.task_id);
let _failure = ResumeDeleteFailure::install(checkpoint_path, crate::heal::DiskError::DiskAccessDenied);
let error = env
@@ -1346,7 +1672,7 @@ mod resume_loop_tests {
.expect_err("checkpoint cleanup failure must fail the erasure-set heal");
assert!(matches!(error, Error::Disk(crate::heal::DiskError::DiskAccessDenied)));
let state = ResumeManager::load_from_disk(env.healer.disk.clone(), "task")
let state = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.expect("completed state must remain discoverable after cleanup failure")
.get_state()
@@ -1354,6 +1680,84 @@ mod resume_loop_tests {
assert!(state.completed, "successful data heal must be persisted before cleanup is attempted");
}
#[tokio::test]
async fn replacement_completion_keeps_resume_artifacts_until_marker_cleanup() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let replacement_task_id = ResumeUtils::generate_task_id();
ResumeManager::new_replacement_intent(
env.healer.disk.clone(),
replacement_task_id.clone(),
"pool_0_set_0".to_string(),
vec!["b".to_string()],
vec!["replacement-a".to_string()],
vec![crate::heal::resume::ReplacementTargetIdentity {
endpoint: "replacement-a".to_string(),
canonical_path: "/mnt/replacement-a".to_string(),
physical_device_ids: vec!["device-a".to_string()],
filesystem_identity: "1:2:3".to_string(),
}],
)
.await
.expect("replacement intent should persist");
let checkpoint = CheckpointManager::new(env.healer.disk.clone(), replacement_task_id.clone())
.await
.expect("replacement checkpoint should persist");
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts::default(),
HealRequestSource::AutoHeal,
)
.with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
healer
.heal_erasure_set(&["b".to_string()], "pool_0_set_0")
.await
.expect("replacement data scan should complete");
let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
.await
.expect("verified replacement state must remain after data scan")
.get_state()
.await;
assert!(state.completed, "the verified state must record a completed data scan");
assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Verified);
assert!(
CheckpointManager::has_checkpoint(&env.healer.disk, &replacement_task_id).await,
"the checkpoint must survive until the caller clears the healing marker"
);
drop(checkpoint);
}
#[tokio::test]
async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() {
let env = make_env().await;
for _ in 0..3 {
assert!(env.resume.schedule_retry().await.expect("retry state should persist"));
env.checkpoint
.reset_for_retry()
.await
.expect("checkpoint reset should persist");
}
env.storage.fail_listing();
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await
.expect_err("exhausted retry state must report the incomplete heal");
assert!(
ResumeManager::has_resume_state(&env.healer.disk, &env.task_id).await,
"retry exhaustion must not delete the resumable state while a marker may remain"
);
assert!(
CheckpointManager::has_checkpoint(&env.healer.disk, &env.task_id).await,
"retry exhaustion must retain the checkpoint with the resumable state"
);
}
#[tokio::test]
async fn retry_resume_repairs_checkpoint_after_crash_between_resets() {
let env = make_env().await;
@@ -1385,7 +1789,7 @@ mod resume_loop_tests {
let (_, checkpoint) = env
.healer
.initialize_resume_state("task", "pool_0_set_0", &["b".to_string()])
.initialize_resume_state(&env.task_id, "pool_0_set_0", &["b".to_string()])
.await
.expect("resume initialization should repair a stale checkpoint");
let checkpoint = checkpoint.get_checkpoint().await;
@@ -1653,4 +2057,132 @@ mod resume_loop_tests {
"the skipped set must be cleared so the retry re-heals the version"
);
}
#[tokio::test]
async fn replacement_target_missing_from_success_result_retries_the_full_pass() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: false,
},
);
env.storage.set_result(
"object",
Some("v1"),
HealResultItem {
after: Infos {
drives: vec![HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "missing".to_string(),
..Default::default()
}],
},
..Default::default()
},
);
let result = env
.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await;
result.expect_err("a missing replacement target must not report completion");
let state = env.resume.get_state().await;
assert!(!state.completed);
assert_eq!(state.retry_count, 1);
assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("v1".to_string()))]);
assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty());
}
#[tokio::test]
async fn replacement_target_readback_evidence_must_confirm_the_healed_version() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts::default(),
HealRequestSource::AutoHeal,
)
.with_replacement_targets(vec!["replacement-a".to_string()], Some("generation-a".to_string()));
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: false,
},
);
env.storage.set_result(
"object",
Some("v1"),
HealResultItem {
after: Infos {
drives: vec![HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "ok".to_string(),
..Default::default()
}],
},
..Default::default()
},
);
env.storage.set_replacement_commit_evidence("object", Some("v1"), false);
let result = healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await;
result.expect_err("a success result without target readback evidence must retry");
assert_eq!(env.resume.get_state().await.retry_count, 1);
assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty());
}
#[tokio::test]
async fn manual_targeted_heal_keeps_existing_best_effort_result_semantics() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts::default(),
HealRequestSource::Admin,
)
.with_replacement_targets(vec!["replacement-a".to_string()], None);
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: false,
},
);
env.storage.set_result(
"object",
Some("v1"),
HealResultItem {
after: Infos {
drives: vec![HealDriveInfo {
endpoint: "replacement-a".to_string(),
state: "ok".to_string(),
..Default::default()
}],
},
..Default::default()
},
);
env.storage.set_replacement_commit_evidence("object", Some("v1"), false);
healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await
.expect("manual targeted healing must retain its existing success semantics");
assert_eq!(env.resume.get_state().await.retry_count, 0);
}
}
File diff suppressed because it is too large Load Diff
+327 -35
View File
@@ -17,6 +17,7 @@ pub mod erasure_healer;
pub mod event;
pub mod manager;
pub mod progress;
pub(crate) mod replacement_readiness;
pub mod resume;
pub mod storage;
pub(crate) mod storage_api;
@@ -25,8 +26,8 @@ pub mod utils;
use storage_api::owner::{
ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore,
EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations,
EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult,
EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations,
ecstore_local_disk_map_read,
};
#[cfg(test)]
@@ -76,56 +77,155 @@ pub(crate) const HEALING_MARKER_PATH: &str = ECSTORE_HEALING_MARKER_PATH;
/// Write the healing marker on the local disks matching `endpoints` so their
/// `DiskInfo.healing` reports true while the erasure-set heal rebuilds them.
pub(crate) async fn set_healing_markers(endpoints: &[String], set_disk_id: &str) {
apply_healing_markers(endpoints, Some(set_disk_id)).await;
pub(crate) async fn set_healing_markers(endpoints: &[String], marker: &str) -> crate::Result<()> {
apply_healing_markers(endpoints, Some(marker), None, false).await
}
/// Remove the healing markers written by [`set_healing_markers`].
pub(crate) async fn clear_healing_markers(endpoints: &[String]) {
apply_healing_markers(endpoints, None).await;
/// Remove an owner marker after the replacement scan's verified state is
/// durable. A missing marker is idempotent here because a crash may have
/// happened after the previous terminal clear and before resume cleanup.
pub(crate) async fn clear_healing_markers_after_verified(endpoints: &[String], marker: &str) -> crate::Result<()> {
apply_healing_markers(endpoints, None, Some(marker), true).await
}
async fn apply_healing_markers(endpoints: &[String], set_disk_id: Option<&str>) {
#[cfg(test)]
fn marker_matches(current: &[u8], expected_marker: Option<&str>) -> bool {
expected_marker.is_some_and(|expected| current == expected.as_bytes())
}
async fn apply_healing_markers(
endpoints: &[String],
marker: Option<&str>,
expected_marker: Option<&str>,
allow_missing: bool,
) -> crate::Result<()> {
if endpoints.is_empty() {
return;
return Ok(());
}
let local_disk_map = local_disk_map_read().await;
for disk in local_disk_map.values().flatten() {
let endpoint = EcstoreDiskAPI::endpoint(disk.as_ref()).to_string();
if !endpoints.iter().any(|candidate| candidate == &endpoint) {
continue;
let mut local_disks = std::collections::HashMap::new();
{
let local_disk_map = local_disk_map_read().await;
for disk in local_disk_map.values().flatten() {
local_disks.insert(EcstoreDiskAPI::endpoint(disk.as_ref()).to_string(), disk.clone());
}
let result = match set_disk_id {
Some(set_disk_id) => {
EcstoreDiskAPI::write_all(
}
let mut matched_endpoints = std::collections::HashSet::new();
let mut targets = Vec::with_capacity(endpoints.len());
for endpoint in endpoints {
if !matched_endpoints.insert(endpoint.clone()) {
return Err(DiskError::other("healing marker endpoint is duplicated").into());
}
let Some(disk) = local_disks.remove(endpoint) else {
return Err(DiskError::other("healing marker target is unavailable").into());
};
targets.push(disk);
}
apply_healing_markers_to_targets(targets, marker, expected_marker, allow_missing).await
}
async fn apply_healing_markers_to_targets(
targets: Vec<DiskStore>,
marker: Option<&str>,
expected_marker: Option<&str>,
allow_missing: bool,
) -> crate::Result<()> {
apply_healing_markers_to_targets_with_after_acquire(targets, marker, expected_marker, allow_missing, |_| {}).await
}
async fn apply_healing_markers_to_targets_with_after_acquire<F>(
targets: Vec<DiskStore>,
marker: Option<&str>,
expected_marker: Option<&str>,
allow_missing: bool,
mut after_acquire: F,
) -> crate::Result<()>
where
F: FnMut(&DiskStore),
{
let marker_bytes = marker.map(|marker| EcstoreDiskBytes::copy_from_slice(marker.as_bytes()));
let expected_bytes = expected_marker.map(|marker| EcstoreDiskBytes::copy_from_slice(marker.as_bytes()));
let mut newly_acquired = Vec::new();
for disk in targets {
let result = match marker_bytes.as_ref() {
Some(marker) => {
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
EcstoreDiskBytes::copy_from_slice(set_disk_id.as_bytes()),
None,
Some(marker.clone()),
)
.await
{
Ok(EcstoreConditionalFileUpdate::Updated) => {
newly_acquired.push(disk.clone());
after_acquire(&disk);
Ok(())
}
Ok(EcstoreConditionalFileUpdate::Mismatch) => match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
Some(marker.clone()),
Some(marker.clone()),
)
.await
{
Ok(EcstoreConditionalFileUpdate::Updated) => Ok(()),
Ok(_) => Err(DiskError::other("healing marker ownership changed")),
Err(err) => Err(err),
},
Ok(EcstoreConditionalFileUpdate::Missing) => Err(DiskError::other("healing marker disappeared")),
Err(err) => Err(err),
}
}
None => {
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
expected_bytes.clone(),
None,
)
.await
{
Ok(EcstoreConditionalFileUpdate::Updated) => Ok(()),
Ok(EcstoreConditionalFileUpdate::Missing) if allow_missing => Ok(()),
Ok(EcstoreConditionalFileUpdate::Missing) => Err(DiskError::other("healing marker is missing")),
Ok(EcstoreConditionalFileUpdate::Mismatch) => Err(DiskError::other("healing marker ownership changed")),
Err(err) => Err(err),
}
}
None => match EcstoreDiskAPI::delete(
disk.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
EcstoreDeleteOptions::default(),
)
.await
{
Err(DiskError::FileNotFound) => Ok(()),
other => other,
},
};
if let Err(err) = result {
tracing::warn!(
endpoint = %endpoint,
action = if set_disk_id.is_some() { "set" } else { "clear" },
error = ?err,
"failed to update healing marker"
);
if let Some(marker) = marker_bytes.as_ref() {
let mut rollback_error = None;
for acquired in newly_acquired.iter().rev() {
if let Err(rollback) = EcstoreDiskAPI::compare_and_update_file(
acquired.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
Some(marker.clone()),
None,
)
.await
{
rollback_error.get_or_insert(rollback);
}
}
if let Some(rollback) = rollback_error {
return Err(DiskError::other(format!(
"healing marker acquisition failed ({err}) and owner-safe rollback failed ({rollback})"
))
.into());
}
}
return Err(err.into());
}
}
Ok(())
}
pub(crate) type DiskError = EcstoreDiskError;
@@ -203,3 +303,195 @@ where
pub type HealObjectInfo = <ECStore as ObjectOperations>::ObjectInfo;
pub type HealObjectOptions = <ECStore as ObjectOperations>::ObjectOptions;
pub type HealPutObjReader = <ECStore as ObjectIO>::PutObjectReader;
#[cfg(test)]
mod tests {
use super::{
DiskError, DiskOption, Endpoint, HEALING_MARKER_PATH, RUSTFS_META_BUCKET, apply_healing_markers_to_targets,
apply_healing_markers_to_targets_with_after_acquire, marker_matches, new_disk,
};
use crate::{
Error,
heal::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes},
};
use tempfile::TempDir;
async fn make_marker_disk(temp: &TempDir, name: &str) -> super::DiskStore {
let path = temp.path().join(name);
std::fs::create_dir_all(&path).expect("marker disk directory should be created");
let endpoint = Endpoint::try_from(path.to_string_lossy().as_ref()).expect("marker disk endpoint should be valid");
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("marker disk should initialize");
let metadata_volume = disk.make_volume(RUSTFS_META_BUCKET).await;
assert!(
matches!(metadata_volume, Ok(()) | Err(DiskError::VolumeExists)),
"marker metadata volume should exist: {metadata_volume:?}"
);
disk
}
#[test]
fn marker_clear_requires_the_current_owner_token() {
assert!(marker_matches(b"set:task-a", Some("set:task-a")));
assert!(!marker_matches(b"set:task-b", Some("set:task-a")));
assert!(!marker_matches(b"set:task-a", None));
}
#[tokio::test]
async fn marker_acquisition_rolls_back_after_second_disk_ownership_conflict() {
let temp = TempDir::new().expect("marker test directory should be created");
let first = make_marker_disk(&temp, "first").await;
let second = make_marker_disk(&temp, "second").await;
let owner_b = EcstoreDiskBytes::from_static(b"owner-b");
assert_eq!(
EcstoreDiskAPI::compare_and_update_file(
second.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
None,
Some(owner_b.clone()),
)
.await
.expect("second disk owner should acquire marker"),
EcstoreConditionalFileUpdate::Updated
);
let err = apply_healing_markers_to_targets(vec![first.clone(), second.clone()], Some("owner-a"), None, false)
.await
.expect_err("second disk ownership must reject the partial acquisition");
assert!(matches!(err, Error::Disk(DiskError::Io(ref io)) if io.to_string() == "healing marker ownership changed"));
assert!(matches!(
EcstoreDiskAPI::read_all(first.as_ref(), RUSTFS_META_BUCKET, HEALING_MARKER_PATH).await,
Err(DiskError::FileNotFound)
));
assert_eq!(
EcstoreDiskAPI::read_all(second.as_ref(), RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.await
.expect("conflicting owner marker must remain"),
owner_b
);
}
#[tokio::test]
async fn marker_acquisition_rolls_back_after_second_disk_io_error() {
let temp = TempDir::new().expect("marker test directory should be created");
let first = make_marker_disk(&temp, "first").await;
let second_path = temp.path().join("second");
std::fs::create_dir_all(&second_path).expect("second marker disk directory should be created");
let second_endpoint =
Endpoint::try_from(second_path.to_string_lossy().as_ref()).expect("second marker endpoint should be valid");
let second = new_disk(
&second_endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("second marker disk should initialize");
std::fs::remove_dir_all(second_path.join(RUSTFS_META_BUCKET))
.expect("second marker metadata directory should be removed for the I/O failure fixture");
std::fs::write(second_path.join(RUSTFS_META_BUCKET), b"not a directory")
.expect("second marker volume should become an I/O failure fixture");
let err = apply_healing_markers_to_targets(vec![first.clone(), second], Some("owner-a"), None, false)
.await
.expect_err("second disk I/O failure must reject the partial acquisition");
assert!(
matches!(err, Error::Disk(DiskError::FileAccessDenied)),
"second marker operation must report its mapped filesystem failure: {err:?}"
);
assert!(matches!(
EcstoreDiskAPI::read_all(first.as_ref(), RUSTFS_META_BUCKET, HEALING_MARKER_PATH).await,
Err(DiskError::FileNotFound)
));
}
#[tokio::test]
async fn marker_acquisition_reports_an_owner_safe_rollback_io_failure() {
let temp = TempDir::new().expect("marker test directory should be created");
let first = make_marker_disk(&temp, "first").await;
let second = make_marker_disk(&temp, "second").await;
let owner_b = EcstoreDiskBytes::from_static(b"owner-b");
let first_path = EcstoreDiskAPI::path(first.as_ref());
let moved_metadata_path = first_path.join("metadata-before-rollback");
assert_eq!(
EcstoreDiskAPI::compare_and_update_file(
second.as_ref(),
RUSTFS_META_BUCKET,
HEALING_MARKER_PATH,
None,
Some(owner_b),
)
.await
.expect("second disk owner should acquire marker"),
EcstoreConditionalFileUpdate::Updated
);
let err =
apply_healing_markers_to_targets_with_after_acquire(vec![first, second], Some("owner-a"), None, false, |disk| {
let metadata_path = EcstoreDiskAPI::path(disk.as_ref()).join(RUSTFS_META_BUCKET);
std::fs::rename(&metadata_path, &moved_metadata_path)
.expect("first marker metadata should move after acquisition");
std::fs::write(&metadata_path, b"not a directory")
.expect("first marker metadata should become a rollback I/O failure fixture");
})
.await
.expect_err("rollback I/O failure must remain visible to the caller");
let message = err.to_string();
assert!(message.contains("healing marker acquisition failed"));
assert!(message.contains("owner-safe rollback failed"));
assert!(moved_metadata_path.join(HEALING_MARKER_PATH).exists());
}
#[tokio::test]
async fn concurrent_marker_acquisition_has_one_owner_on_every_disk() {
let temp = TempDir::new().expect("marker test directory should be created");
let first = make_marker_disk(&temp, "first").await;
let second = make_marker_disk(&temp, "second").await;
let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(3));
let owner_a_barrier = barrier.clone();
let owner_a_first = first.clone();
let owner_a_second = second.clone();
let owner_a = tokio::spawn(async move {
owner_a_barrier.wait().await;
apply_healing_markers_to_targets(vec![owner_a_first, owner_a_second], Some("owner-a"), None, false).await
});
let owner_b_barrier = barrier.clone();
let owner_b_first = first.clone();
let owner_b_second = second.clone();
let owner_b = tokio::spawn(async move {
owner_b_barrier.wait().await;
apply_healing_markers_to_targets(vec![owner_b_first, owner_b_second], Some("owner-b"), None, false).await
});
barrier.wait().await;
let owner_a_result = owner_a.await.expect("owner a task should join");
let owner_b_result = owner_b.await.expect("owner b task should join");
assert_ne!(
owner_a_result.is_ok(),
owner_b_result.is_ok(),
"exactly one owner must acquire both markers"
);
let winning_marker = if owner_a_result.is_ok() { b"owner-a" } else { b"owner-b" };
for disk in [&first, &second] {
assert_eq!(
EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
.await
.expect("every disk must retain the winning owner marker"),
EcstoreDiskBytes::from_static(winning_marker)
);
}
}
}
@@ -0,0 +1,160 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{fs, path::Path};
#[cfg(test)]
use super::Endpoint;
use super::{DiskStore, HealDiskExt as _, local_disk_map_read, 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()
}
pub(crate) async fn auto_replacement_target_identity(
disk: &DiskStore,
local_disks: &[DiskStore],
) -> Option<ReplacementTargetIdentity> {
let lease_root = disk.replacement_mount_lease_root()?;
let endpoint = disk.endpoint().to_string();
let sibling_lease_roots = local_disks
.iter()
.filter(|sibling| sibling.endpoint().is_local && sibling.endpoint().to_string() != endpoint)
.map(|sibling| sibling.replacement_mount_lease_root())
.collect::<Option<Vec<_>>>()?;
tokio::task::spawn_blocking(move || {
let canonical_path = fs::canonicalize(&lease_root).ok()?;
let metadata = fs::metadata(&lease_root).ok()?;
let Ok(target_device_ids) = rustfs_utils::os::get_physical_device_ids(lease_root.to_string_lossy().as_ref()) else {
return None;
};
let Ok(root_device_ids) = rustfs_utils::os::get_physical_device_ids("/") else {
return None;
};
if target_device_ids.is_empty()
|| root_device_ids.is_empty()
|| target_device_ids.iter().any(|target| root_device_ids.contains(target))
|| !rustfs_utils::os::is_mount_point(&canonical_path).unwrap_or(false)
{
return None;
}
if sibling_lease_roots.iter().any(|sibling_lease_root| {
rustfs_utils::os::get_physical_device_ids(sibling_lease_root.to_string_lossy().as_ref())
.map(|ids| ids.iter().any(|id| target_device_ids.contains(id)))
.unwrap_or(true)
}) {
return None;
}
let filesystem_identity = filesystem_identity(&metadata, &canonical_path)?;
Some(ReplacementTargetIdentity {
endpoint,
canonical_path: canonical_path.to_string_lossy().into_owned(),
physical_device_ids: target_device_ids,
filesystem_identity,
})
})
.await
.ok()
.flatten()
}
pub(crate) async fn auto_replacement_targets_ready(targets: &[String]) -> bool {
auto_replacement_target_identities(targets).await.is_some()
}
pub(crate) async fn auto_replacement_target_identities(targets: &[String]) -> Option<Vec<ReplacementTargetIdentity>> {
let local_disk_map = local_disk_map_read().await;
let local_disks = local_disk_map
.values()
.flatten()
.filter(|disk| disk.endpoint().is_local)
.cloned()
.collect::<Vec<_>>();
drop(local_disk_map);
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?);
}
identities.sort_by(|left, right| left.endpoint.cmp(&right.endpoint));
identities.dedup_by(|left, right| left.endpoint == right.endpoint);
(identities.len() == targets.len()).then_some(identities)
}
#[cfg(target_os = "linux")]
fn filesystem_identity(metadata: &fs::Metadata, canonical_path: &Path) -> Option<String> {
use std::os::unix::fs::MetadataExt as _;
let escaped_path = canonical_path.to_string_lossy().replace(' ', "\\040");
let mountinfo = fs::read_to_string("/proc/self/mountinfo").ok()?;
let mount_id = mountinfo.lines().find_map(|line| {
let mut fields = line.split_whitespace();
let mount_id = fields.next()?;
fields.next()?;
fields.next()?;
fields.next()?;
(fields.next()? == escaped_path).then_some(mount_id)
})?;
Some(format!("{mount_id}:{}:{}", metadata.dev(), metadata.ino()))
}
#[cfg(all(unix, not(target_os = "linux")))]
fn filesystem_identity(metadata: &fs::Metadata, _canonical_path: &Path) -> Option<String> {
use std::os::unix::fs::MetadataExt as _;
Some(format!("{}:{}", metadata.dev(), metadata.ino()))
}
#[cfg(not(unix))]
fn filesystem_identity(_metadata: &fs::Metadata, _canonical_path: &Path) -> Option<String> {
None
}
#[cfg(test)]
mod tests {
use super::super::{DiskOption, new_disk};
use super::*;
use tempfile::TempDir;
#[tokio::test]
async fn runtime_environment_cannot_bypass_mount_admission() {
temp_env::async_with_vars(
[
("RUSTFS_TEST_AUTO_REPLACEMENT_READINESS_BYPASS", Some("1")),
("RUSTFS_E2E_AUTO_REPLACEMENT_READINESS_BYPASS", Some("1")),
],
async {
let temp = TempDir::new().expect("temporary replacement root should be created");
let endpoint =
Endpoint::try_from(temp.path().to_string_lossy().as_ref()).expect("replacement endpoint should parse");
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("temporary disk should initialize");
assert!(!auto_replacement_target_ready(&disk, std::slice::from_ref(&disk)).await);
},
)
.await;
}
}
File diff suppressed because it is too large Load Diff
+156 -4
View File
@@ -26,7 +26,7 @@ use super::storage_api::storage::{
BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _,
ObjectOperations as _, StorageAdminApi,
};
use super::{DiskStore, ECStore, Endpoint, StorageError};
use super::{DiskStore, ECStore, Endpoint, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity};
pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
const LOG_COMPONENT_HEAL: &str = "heal";
@@ -37,6 +37,11 @@ const EVENT_HEAL_STORAGE_OBJECT_VERIFY: &str = "heal_storage_object_verify";
const EVENT_HEAL_STORAGE_ADMIN_OP: &str = "heal_storage_admin_op";
const EVENT_HEAL_STORAGE_REPAIR_OP: &str = "heal_storage_repair_op";
pub enum ReplacementResumeDisk {
Fresh,
Existing(DiskStore),
}
pub(crate) fn next_heal_listing_token(
bucket: &str,
prefix: &str,
@@ -354,6 +359,42 @@ pub trait HealStorageAPI: Send + Sync {
/// Heal format using ecstore
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)>;
/// Heal only the explicitly admitted replacement targets in one erasure set.
///
/// The default is deliberately fail-closed so alternate storage
/// implementations cannot accidentally fall back to the global format path.
async fn heal_replacement_format(
&self,
_dry_run: bool,
_pool_index: usize,
_set_index: usize,
_targets: &[String],
) -> Result<(HealResultItem, Option<Error>)> {
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
/// transaction returns success. The conservative default prevents an
/// alternate backend from turning an unverified replacement into a
/// completed generation.
async fn replacement_targets_have_version(
&self,
_bucket: &str,
_object: &str,
_version_id: Option<&str>,
_opts: &HealOpts,
_targets: &[String],
) -> Result<bool> {
Ok(false)
}
/// List object versions for healing (returns all versions, may use significant memory for large buckets)
///
/// WARNING: This method loads all object versions into memory at once. For buckets with many
@@ -390,8 +431,30 @@ pub trait HealStorageAPI: Send + Sync {
self.list_objects_for_heal_page(bucket, prefix, continuation_token).await
}
/// Get disk for resume functionality
/// Get disk for resume functionality.
async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore>;
/// Get a healthy non-target disk for durable replacement state.
async fn get_disk_for_resume_excluding(&self, _set_disk_id: &str, _excluded_targets: &[String]) -> Result<DiskStore> {
Err(Error::other("target-excluding resume disk selection is unsupported"))
}
/// Reopen the exact surviving disk that owns an existing replacement
/// intent. Falling back to another disk would create a second copy of the
/// same generation and split its progress.
async fn get_replacement_resume_disk(
&self,
_set_disk_id: &str,
_task_id: &str,
_excluded_targets: &[String],
) -> Result<ReplacementResumeDisk> {
Err(Error::other("durable replacement resume selection is unsupported"))
}
/// Capture the mounted replacement instance before it is formatted.
async fn replacement_target_identities(&self, _targets: &[String]) -> Result<Vec<ReplacementTargetIdentity>> {
Err(Error::other("replacement target identity collection is unsupported"))
}
}
/// ECStore Heal storage layer implementation
@@ -1306,6 +1369,44 @@ impl HealStorageAPI for ECStoreHealStorage {
}
}
async fn heal_replacement_format(
&self,
dry_run: bool,
pool_index: usize,
set_index: usize,
targets: &[String],
) -> Result<(HealResultItem, Option<Error>)> {
self.ecstore
.heal_replacement_format(dry_run, pool_index, set_index, targets)
.await
.map(|(result, error)| (result, error.map(Error::Storage)))
.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,
object: &str,
version_id: Option<&str>,
opts: &HealOpts,
targets: &[String],
) -> Result<bool> {
let pool_index = opts
.pool
.ok_or_else(|| Error::other("replacement target readback is missing pool scope"))?;
let set_index = opts
.set
.ok_or_else(|| Error::other("replacement target readback is missing set scope"))?;
self.ecstore
.replacement_targets_have_version(bucket, object, version_id.unwrap_or(""), pool_index, set_index, targets)
.await
.map_err(Error::Storage)
}
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>> {
debug!(
target: "rustfs::heal::storage",
@@ -1543,6 +1644,10 @@ impl HealStorageAPI for ECStoreHealStorage {
}
async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore> {
self.get_disk_for_resume_excluding(set_disk_id, &[]).await
}
async fn get_disk_for_resume_excluding(&self, set_disk_id: &str, excluded_targets: &[String]) -> Result<DiskStore> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1564,8 +1669,18 @@ impl HealStorageAPI for ECStoreHealStorage {
message: format!("Failed to get disks for pool {pool_idx} set {set_idx}: {e}"),
})?;
// Find the first available disk
if let Some(disk_store) = disks.into_iter().flatten().next() {
// The replacement target is unformatted before repair and must never
// host the intent that authorizes its own formatting.
for disk_store in disks.into_iter().flatten() {
if !disk_store.endpoint().is_local {
continue;
}
if excluded_targets.contains(&disk_store.endpoint().to_string()) {
continue;
}
if !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) {
continue;
}
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1584,6 +1699,43 @@ impl HealStorageAPI for ECStoreHealStorage {
message: format!("No available disk found for set_disk_id: {set_disk_id}"),
})
}
async fn get_replacement_resume_disk(
&self,
set_disk_id: &str,
task_id: &str,
excluded_targets: &[String],
) -> Result<ReplacementResumeDisk> {
let (pool_idx, set_idx) = crate::heal::utils::parse_set_disk_id(set_disk_id)?;
let disks = StorageAdminApi::disk_set_inventory(self.ecstore.as_ref(), DiskSetSelector::new(pool_idx, set_idx))
.await
.map_err(|e| Error::TaskExecutionFailed {
message: format!("Failed to get disks for pool {pool_idx} set {set_idx}: {e}"),
})?;
let mut existing = None;
for disk_store in disks.into_iter().flatten() {
if !disk_store.endpoint().is_local || excluded_targets.contains(&disk_store.endpoint().to_string()) {
continue;
}
if !matches!(disk_store.get_disk_id().await, Ok(Some(id)) if !id.is_nil()) {
continue;
}
if super::resume::ResumeManager::has_replacement_intent(&disk_store, task_id).await
&& existing.replace(disk_store).is_some()
{
return Err(Error::TaskExecutionFailed {
message: format!("Replacement resume intent is duplicated for set_disk_id: {set_disk_id}"),
});
}
}
Ok(existing.map_or(ReplacementResumeDisk::Fresh, ReplacementResumeDisk::Existing))
}
async fn replacement_target_identities(&self, targets: &[String]) -> Result<Vec<ReplacementTargetIdentity>> {
super::replacement_readiness::auto_replacement_target_identities(targets)
.await
.ok_or_else(|| Error::other("replacement target is not a stable mounted disk"))
}
}
#[cfg(test)]
+6 -4
View File
@@ -16,8 +16,9 @@ pub(crate) use rustfs_ecstore::api::data_usage::DATA_USAGE_CACHE_NAME as ECSTORE
pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint;
pub(crate) use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult};
pub(crate) use rustfs_ecstore::api::disk::{
BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes, DeleteOptions as EcstoreDeleteOptions,
DiskAPI as EcstoreDiskAPI, DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH,
BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes,
ConditionalFileUpdate as EcstoreConditionalFileUpdate, DeleteOptions as EcstoreDeleteOptions, DiskAPI as EcstoreDiskAPI,
DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH,
RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET,
};
#[cfg(test)]
@@ -32,8 +33,9 @@ pub(crate) mod owner {
pub(crate) use super::{
ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore,
EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ecstore_local_disk_map_read,
EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError,
EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore,
ecstore_local_disk_map_read,
};
#[cfg(test)]
+672 -12
View File
@@ -13,8 +13,12 @@
// limitations under the License.
use crate::heal::{
DiskError, EcstoreError, ErasureSetHealer,
DiskError, EcstoreError, ErasureSetHealer, HealDiskExt as _,
erasure_healer::target_outcomes_complete,
progress::HealProgress,
resume::{
CheckpointManager, ReplacementPhase, ReplacementTargetIdentity, ResumeManager, replacement_target_identities_match,
},
storage::{HealStorageAPI, next_heal_listing_token},
};
use crate::{Error, Result};
@@ -350,6 +354,9 @@ pub struct HealTask {
pub retry_attempts: u32,
/// Endpoints of the disks being rebuilt (see `HealRequest::heal_endpoints`).
pub heal_endpoints: Vec<String>,
/// Durable resume anchor injected by the manager for an existing automatic
/// replacement generation.
replacement_resume_endpoint: Option<String>,
/// Task status
pub status: Arc<RwLock<HealTaskStatus>>,
/// Progress tracking
@@ -376,6 +383,24 @@ pub struct HealTask {
}
impl HealTask {
async fn verify_replacement_identity_fence(
&self,
expected_identities: &[ReplacementTargetIdentity],
set_disk_id: &str,
stage: &str,
) -> Result<()> {
let actual_identities = self
.await_with_control(self.storage.replacement_target_identities(&self.heal_endpoints))
.await?;
if replacement_target_identities_match(expected_identities, &actual_identities) {
return Ok(());
}
Err(Error::TaskExecutionFailed {
message: format!("Replacement target changed during {stage} for automatic heal {set_disk_id}"),
})
}
pub fn from_request(request: HealRequest, storage: Arc<dyn HealStorageAPI>) -> Self {
Self {
id: request.id,
@@ -385,6 +410,7 @@ impl HealTask {
source: request.source,
retry_attempts: request.retry_attempts,
heal_endpoints: request.heal_endpoints,
replacement_resume_endpoint: None,
status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
progress: Arc::new(RwLock::new(HealProgress::new())),
result_items: Arc::new(RwLock::new(Vec::new())),
@@ -416,6 +442,16 @@ impl HealTask {
}
}
pub(crate) fn from_replacement_recovery_request(
request: HealRequest,
storage: Arc<dyn HealStorageAPI>,
replacement_resume_endpoint: Option<String>,
) -> Self {
let mut task = Self::from_request(request, storage);
task.replacement_resume_endpoint = replacement_resume_endpoint;
task
}
pub fn metric_type_label(&self) -> &'static str {
match &self.heal_type {
HealType::Cluster => "cluster",
@@ -2131,7 +2167,87 @@ impl HealTask {
progress.update_progress(0, 4, 0, 0);
}
let buckets = if buckets.is_empty() {
let is_auto_replacement = matches!(self.source, HealRequestSource::AutoHeal) && !self.heal_endpoints.is_empty();
let replacement_resume_disk = if is_auto_replacement {
let mut requested_targets = self.heal_endpoints.clone();
requested_targets.sort_unstable();
requested_targets.dedup();
let selection = self
.await_with_control(
self.storage
.get_replacement_resume_disk(&set_disk_id, &self.id, &self.heal_endpoints),
)
.await?;
let disk = match selection {
crate::heal::storage::ReplacementResumeDisk::Existing(disk) => {
if let Some(anchor) = &self.replacement_resume_endpoint
&& disk.endpoint().to_string() != *anchor
{
return Err(Error::TaskExecutionFailed {
message: format!("Replacement resume anchor changed for automatic heal {set_disk_id}"),
});
}
Some(disk)
}
crate::heal::storage::ReplacementResumeDisk::Fresh => {
if self.replacement_resume_endpoint.is_some() {
return Err(Error::TaskExecutionFailed {
message: format!("Replacement resume anchor is unavailable for automatic heal {set_disk_id}"),
});
}
None
}
};
if let Some(disk) = disk.as_ref()
&& ResumeManager::has_replacement_intent(disk, &self.id).await
{
let resume_manager = ResumeManager::load_replacement_intent(disk.clone(), &self.id).await?;
let state = resume_manager.get_state().await;
if state.completed
&& matches!(state.replacement_phase, ReplacementPhase::CleanupPending)
&& state.set_disk_id == set_disk_id
&& state.replacement_targets == requested_targets
&& state.replacement_generation.as_deref() == Some(self.id.as_str())
{
resume_manager.ensure_replacement_completion_proof().await?;
if CheckpointManager::has_checkpoint(disk, &self.id).await {
CheckpointManager::load_from_disk(disk.clone(), &self.id)
.await?
.cleanup()
.await?;
}
resume_manager.cleanup().await?;
return Ok(());
}
}
disk
} else {
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,
None => {
self.await_with_control(self.storage.get_disk_for_resume_excluding(&set_disk_id, &self.heal_endpoints))
.await?
}
})
} else {
None
};
let mut buckets = if buckets.is_empty() {
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_STAGE,
@@ -2148,6 +2264,50 @@ impl HealTask {
buckets
};
// Persist automatic replacement intent on a surviving disk before the
// first target format write. A task retry keeps this id; a newly
// admitted blank replacement gets a fresh id and cannot reuse cursor
// progress from an older disk at the same endpoint.
let replacement_resume = if is_auto_replacement {
let identities = self
.await_with_control(self.storage.replacement_target_identities(&self.heal_endpoints))
.await?;
let disk = replacement_resume_disk.clone().ok_or_else(|| Error::TaskExecutionFailed {
message: format!("Replacement resume disk is missing for automatic heal {set_disk_id}"),
})?;
let manager = ResumeManager::new_replacement_intent(
disk.clone(),
self.id.clone(),
set_disk_id.clone(),
buckets.clone(),
self.heal_endpoints.clone(),
identities.clone(),
)
.await?;
buckets = manager.get_state().await.replacement_buckets;
Some((disk, manager, identities))
} else {
None
};
let healing_marker = format!("{set_disk_id}:{}", self.id);
if let Some((disk, resume_manager, _)) = replacement_resume.as_ref() {
let state = resume_manager.get_state().await;
if state.completed && matches!(state.replacement_phase, ReplacementPhase::Verified) {
resume_manager.ensure_replacement_completion_proof().await?;
super::clear_healing_markers_after_verified(&self.heal_endpoints, &healing_marker).await?;
resume_manager.mark_replacement_cleanup_pending().await?;
if CheckpointManager::has_checkpoint(disk, &self.id).await {
CheckpointManager::load_from_disk(disk.clone(), &self.id)
.await?
.cleanup()
.await?;
}
resume_manager.cleanup().await?;
return Ok(());
}
}
// Step 1: Perform disk format heal using ecstore
debug!(
target: "rustfs::heal::task",
@@ -2159,7 +2319,32 @@ impl HealTask {
stage = "heal_format",
"Heal erasure set stage entered"
);
let format_result = self.await_with_control(self.storage.heal_format(self.options.dry_run)).await;
if is_auto_replacement {
let Some((_, _, expected_identities)) = replacement_resume.as_ref() else {
return Err(Error::TaskExecutionFailed {
message: format!("Replacement intent is missing for automatic heal {set_disk_id}"),
});
};
self.verify_replacement_identity_fence(expected_identities, &set_disk_id, "format")
.await?;
}
let format_result = if is_auto_replacement {
let pool_index = self.options.pool_index.ok_or_else(|| Error::TaskExecutionFailed {
message: format!("Missing pool scope for automatic replacement heal {set_disk_id}"),
})?;
let set_index = self.options.set_index.ok_or_else(|| Error::TaskExecutionFailed {
message: format!("Missing set scope for automatic replacement heal {set_disk_id}"),
})?;
self.await_with_control(self.storage.heal_replacement_format(
self.options.dry_run,
pool_index,
set_index,
&self.heal_endpoints,
))
.await
} else {
self.await_with_control(self.storage.heal_format(self.options.dry_run)).await
};
match format_result {
Ok((result, error)) => {
@@ -2209,6 +2394,22 @@ impl HealTask {
"Heal erasure set format repaired"
);
}
if !self.options.dry_run && !target_outcomes_complete(&result, &self.heal_endpoints) {
return Err(Error::TaskExecutionFailed {
message: format!("Failed to verify formatted replacement targets for {set_disk_id}"),
});
}
if let Some((_, replacement_resume, expected_identities)) = &replacement_resume {
let identities = self
.await_with_control(self.storage.replacement_target_identities(&self.heal_endpoints))
.await?;
if !replacement_target_identities_match(expected_identities, &identities) {
return Err(Error::TaskExecutionFailed {
message: format!("Replacement target changed after format for automatic heal {set_disk_id}"),
});
}
replacement_resume.mark_replacement_rebuilding(identities).await?;
}
}
Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
@@ -2241,7 +2442,7 @@ impl HealTask {
// The rebuilt disks are formatted now: mark them as healing so
// DiskInfo.healing reflects the rebuild until it completes.
super::set_healing_markers(&self.heal_endpoints, &set_disk_id).await;
super::set_healing_markers(&self.heal_endpoints, &healing_marker).await?;
// Step 2: Get disk for resume functionality
debug!(
@@ -2254,9 +2455,14 @@ impl HealTask {
stage = "resolve_resume_disk",
"Heal erasure set stage entered"
);
let disk = self
.await_with_control(self.storage.get_disk_for_resume(&set_disk_id))
.await?;
let replacement_target_identities = replacement_resume.as_ref().map(|(_, _, identities)| identities.clone());
let disk = match replacement_resume.as_ref() {
Some((disk, _, _)) => disk.clone(),
None => {
self.await_with_control(self.storage.get_disk_for_resume(&set_disk_id))
.await?
}
};
{
let mut progress = self.progress.write().await;
@@ -2280,6 +2486,10 @@ impl HealTask {
for bucket in buckets.iter() {
// Check control flags before starting each bucket heal
self.check_control_flags().await?;
if let Some(expected_identities) = replacement_target_identities.as_ref() {
self.verify_replacement_identity_fence(expected_identities, &set_disk_id, "bucket prepass")
.await?;
}
let heal_result = self
.await_with_control(self.storage.heal_bucket(bucket, &bucket_heal_opts))
.await;
@@ -2334,7 +2544,9 @@ impl HealTask {
disk,
heal_opts,
self.source,
);
)
.with_replacement_targets(self.heal_endpoints.clone(), is_auto_replacement.then(|| self.id.clone()))
.with_replacement_identity_fence(replacement_target_identities.clone());
{
let mut progress = self.progress.write().await;
@@ -2358,9 +2570,27 @@ impl HealTask {
// Keep the markers on failure: the resume state also persists, and the
// next run of this set heal re-marks and eventually clears them.
if result.is_ok() {
super::clear_healing_markers(&self.heal_endpoints).await;
}
let result = match result {
Ok(()) => {
if let Some(expected_identities) = replacement_target_identities.as_ref() {
self.verify_replacement_identity_fence(expected_identities, &set_disk_id, "marker completion")
.await?;
}
super::clear_healing_markers_after_verified(&self.heal_endpoints, &healing_marker).await?;
if let Some((disk, resume_manager, _)) = replacement_resume.as_ref() {
resume_manager.mark_replacement_cleanup_pending().await?;
if CheckpointManager::has_checkpoint(disk, &self.id).await {
CheckpointManager::load_from_disk(disk.clone(), &self.id)
.await?
.cleanup()
.await?;
}
resume_manager.cleanup().await?;
}
Ok(())
}
Err(err) => Err(err),
};
{
let mut progress = self.progress.write().await;
@@ -2420,12 +2650,353 @@ mod tests {
use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk};
use super::*;
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo};
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos};
use std::collections::{HashMap, VecDeque};
use std::sync::Mutex;
use tempfile::TempDir;
use super::super::storage_api::status::BucketInfo;
#[test]
fn format_result_requires_every_requested_target_to_be_ok() {
let result = HealResultItem {
after: Infos {
drives: vec![
HealDriveInfo {
endpoint: "disk-a".to_string(),
state: "ok".to_string(),
..Default::default()
},
HealDriveInfo {
endpoint: "disk-b".to_string(),
state: "missing".to_string(),
..Default::default()
},
],
},
..Default::default()
};
assert!(target_outcomes_complete(&result, &["disk-a".to_string()]));
assert!(!target_outcomes_complete(&result, &["disk-a".to_string(), "disk-b".to_string()]));
assert!(!target_outcomes_complete(&result, &["disk-c".to_string()]));
}
#[tokio::test]
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),
resume_disk: Mutex::new(Some(disk)),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: Vec::new(),
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..Default::default()
},
HealPriority::Low,
);
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
let task = HealTask::from_request(request, storage.clone());
task.execute()
.await
.expect_err("the mock has no local replacement marker target");
assert_eq!(
*storage.global_format_calls.lock().unwrap(),
0,
"automatic replacement must not call global format"
);
assert_eq!(
storage.replacement_format_calls.lock().unwrap().as_slice(),
&[(0, 0, vec!["replacement-a".to_string()])],
"automatic replacement must pass the exact pool, set, and target"
);
}
#[tokio::test]
async fn automatic_replacement_persists_intent_before_format() {
let storage = Arc::new(MockStorage {
replacement_targets_ready: Mutex::new(true),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..HealOptions::default()
},
HealPriority::Low,
);
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
HealTask::from_request(request, storage.clone())
.execute()
.await
.expect_err("intent persistence needs a healthy non-target disk");
assert!(
storage.replacement_format_calls.lock().unwrap().is_empty(),
"format must not start before the durable replacement intent exists"
);
}
#[tokio::test]
async fn recovered_replacement_never_uses_a_fresh_resume_disk() {
let storage = Arc::new(MockStorage {
replacement_targets_ready: Mutex::new(true),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..HealOptions::default()
},
HealPriority::Low,
);
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
let error = HealTask::from_replacement_recovery_request(request, storage.clone(), Some("survivor-a".to_string()))
.execute()
.await
.expect_err("a durable recovery must not fall back to another resume disk");
assert!(error.to_string().contains("resume anchor is unavailable"));
assert!(
storage.replacement_format_calls.lock().unwrap().is_empty(),
"an unavailable durable anchor must block formatting before any write"
);
assert!(!*storage.listed.lock().unwrap(), "an unavailable durable anchor must not list buckets");
}
#[tokio::test]
async fn automatic_replacement_rejects_a_new_identity_after_format() {
let temp = TempDir::new().expect("temporary resume disk directory should be created");
let disk = make_resume_disk(&temp).await;
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_identity_sequences: Mutex::new(VecDeque::from([
vec![first_identity.clone()],
vec![first_identity.clone()],
vec![second_identity],
])),
resume_disk: Mutex::new(Some(disk.clone())),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..HealOptions::default()
},
HealPriority::Low,
);
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
let task = HealTask::from_request(request, storage.clone());
let error = task
.execute()
.await
.expect_err("a remounted target after format must fail closed");
assert!(error.to_string().contains("changed after format"));
assert_eq!(storage.replacement_format_calls.lock().unwrap().len(), 1);
assert!(storage.bucket_heal_calls.lock().unwrap().is_empty());
assert!(storage.heal_object_calls.lock().unwrap().is_empty());
let state = ResumeManager::load_replacement_intent(disk, &task.id)
.await
.expect("durable replacement intent should remain available")
.get_state()
.await;
assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Intent);
assert_eq!(state.replacement_target_identities, vec![first_identity]);
}
#[tokio::test]
async fn automatic_replacement_reuses_an_existing_non_target_resume_anchor() {
let temp = TempDir::new().expect("temporary resume disk directory should be created");
let anchor = make_resume_disk(&temp).await;
let task_id = crate::heal::resume::ResumeUtils::generate_task_id();
let identity = ReplacementTargetIdentity {
endpoint: "replacement-a".to_string(),
canonical_path: "/replacement/replacement-a".to_string(),
physical_device_ids: vec!["replacement-a".to_string()],
filesystem_identity: "identity-replacement-a".to_string(),
};
ResumeManager::new_replacement_intent(
anchor.clone(),
task_id.clone(),
"pool_0_set_0".to_string(),
vec!["bucket-a".to_string()],
vec!["replacement-a".to_string()],
vec![identity],
)
.await
.expect("existing intent should be stored on the non-target anchor");
let storage = Arc::new(MockStorage {
replacement_targets_ready: Mutex::new(true),
replacement_resume_disk: Mutex::new(Some(anchor.clone())),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..HealOptions::default()
},
HealPriority::Low,
);
request.id = task_id.clone();
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
HealTask::from_request(request, storage.clone())
.execute()
.await
.expect_err("the test has no mounted marker target after format");
assert_eq!(
storage.replacement_format_calls.lock().unwrap().len(),
1,
"an existing non-target anchor must be reused instead of falling back to a fresh anchor"
);
assert!(
storage.resume_disk.lock().unwrap().is_none(),
"the fresh resume-anchor fallback must remain unused"
);
let state = ResumeManager::load_replacement_intent(anchor, &task_id)
.await
.expect("the existing non-target anchor should retain the generation")
.get_state()
.await;
assert_eq!(state.replacement_phase, ReplacementPhase::Rebuilding);
}
#[tokio::test]
async fn automatic_replacement_defers_before_bucket_listing_when_target_is_unready() {
let storage = Arc::new(MockStorage::default());
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: Vec::new(),
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..Default::default()
},
HealPriority::Low,
);
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
HealTask::from_request(request, storage.clone())
.execute()
.await
.expect_err("an unsafe replacement must defer before any scan work");
assert!(!*storage.listed.lock().unwrap(), "unsafe targets must not list buckets");
assert!(
storage.replacement_format_calls.lock().unwrap().is_empty(),
"unsafe targets must not format"
);
}
#[tokio::test]
async fn cleanup_pending_recovery_skips_target_readiness_and_format() {
let temp = TempDir::new().expect("temporary resume disk directory should be created");
let anchor = make_resume_disk(&temp).await;
let task_id = crate::heal::resume::ResumeUtils::generate_task_id();
let identity = replacement_identity("replacement-a", "device-a", "filesystem-a");
let resume_manager = ResumeManager::new_replacement_intent(
anchor.clone(),
task_id.clone(),
"pool_0_set_0".to_string(),
vec!["bucket-a".to_string()],
vec!["replacement-a".to_string()],
vec![identity],
)
.await
.expect("terminal replacement state should persist on the survivor anchor");
resume_manager
.mark_replacement_completed_and_verified()
.await
.expect("terminal replacement proof should persist before cleanup");
resume_manager
.mark_replacement_cleanup_pending()
.await
.expect("failed cleanup must retain a cleanup-pending state");
let storage = Arc::new(MockStorage {
replacement_resume_disk: Mutex::new(Some(anchor.clone())),
..Default::default()
});
let mut request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
pool_index: Some(0),
set_index: Some(0),
..HealOptions::default()
},
HealPriority::Low,
);
request.id = task_id.clone();
request.source = HealRequestSource::AutoHeal;
request.heal_endpoints = vec!["replacement-a".to_string()];
HealTask::from_replacement_recovery_request(request, storage.clone(), Some(anchor.endpoint().to_string()))
.execute()
.await
.expect("cleanup-pending recovery must not require a mounted replacement target");
assert!(
!ResumeManager::has_resume_state(&anchor, &task_id).await,
"terminal cleanup must remove the retained resume state"
);
assert_eq!(*storage.global_format_calls.lock().unwrap(), 0);
assert!(
storage.replacement_format_calls.lock().unwrap().is_empty(),
"terminal cleanup must not format replacement targets"
);
assert!(storage.bucket_heal_calls.lock().unwrap().is_empty());
assert!(!*storage.listed.lock().unwrap());
}
#[derive(Default)]
struct MockStorage {
listed: Mutex<bool>,
@@ -2440,6 +3011,10 @@ mod tests {
heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
deleted_objects: Mutex<Vec<String>>,
format_no_heal_required: Mutex<bool>,
global_format_calls: Mutex<u32>,
replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
replacement_targets_ready: Mutex<bool>,
replacement_target_identity_sequences: Mutex<VecDeque<Vec<crate::heal::resume::ReplacementTargetIdentity>>>,
listed_prefixes: Mutex<Vec<String>>,
truncate_without_token: Mutex<bool>,
include_object_dir_candidate: Mutex<bool>,
@@ -2448,6 +3023,7 @@ mod tests {
bucket_heal_calls: Mutex<Vec<String>>,
block_heal_object: Mutex<bool>,
resume_disk: Mutex<Option<DiskStore>>,
replacement_resume_disk: Mutex<Option<DiskStore>>,
}
#[test]
@@ -2519,6 +3095,19 @@ mod tests {
}
}
fn replacement_identity(
endpoint: &str,
physical_device_id: &str,
filesystem_identity: &str,
) -> crate::heal::resume::ReplacementTargetIdentity {
crate::heal::resume::ReplacementTargetIdentity {
endpoint: endpoint.to_string(),
canonical_path: format!("/replacement/{endpoint}"),
physical_device_ids: vec![physical_device_id.to_string()],
filesystem_identity: filesystem_identity.to_string(),
}
}
enum MockHealObjectOutcome {
OkWithOtherError(&'static str),
ErrOther(&'static str),
@@ -2721,6 +3310,7 @@ mod tests {
}
async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
*self.global_format_calls.lock().unwrap() += 1;
let no_heal_required = *self.format_no_heal_required.lock().unwrap();
if no_heal_required {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2729,6 +3319,39 @@ mod tests {
}
}
async fn heal_replacement_format(
&self,
_dry_run: bool,
pool_index: usize,
set_index: usize,
targets: &[String],
) -> Result<(HealResultItem, Option<Error>)> {
self.replacement_format_calls
.lock()
.unwrap()
.push((pool_index, set_index, targets.to_vec()));
Ok((
HealResultItem {
after: Infos {
drives: targets
.iter()
.map(|endpoint| HealDriveInfo {
endpoint: endpoint.clone(),
state: "ok".to_string(),
..Default::default()
})
.collect(),
},
..Default::default()
},
None,
))
}
async fn replacement_targets_ready(&self, _targets: &[String]) -> Result<bool> {
Ok(*self.replacement_targets_ready.lock().unwrap())
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<HealListItem>> {
Ok(vec![heal_item("object-a"), heal_item("object-b")])
}
@@ -2772,6 +3395,43 @@ mod tests {
.clone()
.ok_or_else(|| Error::other("not implemented in tests"))
}
async fn get_disk_for_resume_excluding(&self, set_disk_id: &str, _excluded_targets: &[String]) -> Result<DiskStore> {
self.get_disk_for_resume(set_disk_id).await
}
async fn get_replacement_resume_disk(
&self,
_set_disk_id: &str,
_task_id: &str,
_excluded_targets: &[String],
) -> Result<crate::heal::storage::ReplacementResumeDisk> {
if let Some(disk) = self.replacement_resume_disk.lock().unwrap().clone() {
return Ok(crate::heal::storage::ReplacementResumeDisk::Existing(disk));
}
Ok(crate::heal::storage::ReplacementResumeDisk::Fresh)
}
async fn replacement_target_identities(
&self,
targets: &[String],
) -> Result<Vec<crate::heal::resume::ReplacementTargetIdentity>> {
if !*self.replacement_targets_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() {
return Ok(identities);
}
Ok(targets
.iter()
.map(|endpoint| crate::heal::resume::ReplacementTargetIdentity {
endpoint: endpoint.clone(),
canonical_path: format!("/replacement/{endpoint}"),
physical_device_ids: vec![endpoint.clone()],
filesystem_identity: format!("identity-{endpoint}"),
})
.collect())
}
}
#[tokio::test]
+90 -1
View File
@@ -18,9 +18,12 @@ pub mod heal;
pub use error::{Error, Result};
pub use heal::{
HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType,
channel::HealChannelProcessor, progress::HealProgress,
channel::HealChannelProcessor,
progress::HealProgress,
resume::{ReplacementRecoveryRecord, ReplacementRecoveryState, ResumeUtils},
};
use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
use std::collections::BTreeMap;
use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
@@ -73,6 +76,17 @@ static GLOBAL_HEAL_RUNTIME_INIT: Mutex<()> = Mutex::const_new(());
static GLOBAL_HEAL_ACTIVE_TASKS: AtomicU64 = AtomicU64::new(0);
static GLOBAL_HEAL_QUEUE_LENGTH: AtomicU64 = AtomicU64::new(0);
/// Local view of durable replacement recovery state. `definitive` only covers
/// the local survivor-disk records; a distributed caller must additionally
/// establish that every peer returned a compatible snapshot.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ReplacementRecoverySnapshot {
pub records: Vec<ReplacementRecoveryRecord>,
pub definitive: bool,
pub reason: Option<String>,
}
#[cfg(test)]
#[derive(Default)]
struct HealRuntimeInitTestHook {
@@ -243,6 +257,81 @@ pub async fn current_heal_progress_snapshot() -> Option<HealProgress> {
}
}
/// Read all local survivor-disk replacement records without conflating an I/O
/// failure or conflicting copies with successful completion.
pub async fn current_replacement_recovery_snapshot() -> ReplacementRecoverySnapshot {
if !heal_runtime_initialized() {
return ReplacementRecoverySnapshot {
records: Vec::new(),
definitive: false,
reason: Some("heal runtime is not initialized".to_string()),
};
}
let disks = {
let local_disk_map = heal::local_disk_map_read().await;
local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
};
if disks.is_empty() {
return ReplacementRecoverySnapshot {
records: Vec::new(),
definitive: false,
reason: Some("no local survivor disks are available".to_string()),
};
}
let mut records = BTreeMap::<String, ReplacementRecoveryRecord>::new();
let mut reason = None;
for disk in disks {
match ResumeUtils::get_replacement_recovery_records(&disk).await {
Ok(disk_records) => {
for record in disk_records {
let task_id = record.task_id.clone();
if matches!(record.state, ReplacementRecoveryState::Unknown) {
reason.get_or_insert_with(|| "invalid durable replacement record".to_string());
}
match records.entry(task_id.clone()) {
std::collections::btree_map::Entry::Vacant(entry) => {
entry.insert(record);
}
std::collections::btree_map::Entry::Occupied(entry) if entry.get() == &record => {}
std::collections::btree_map::Entry::Occupied(entry)
if matches!(entry.get().state, ReplacementRecoveryState::CleanupPending)
&& matches!(record.state, ReplacementRecoveryState::Completed) => {}
std::collections::btree_map::Entry::Occupied(mut entry)
if matches!(entry.get().state, ReplacementRecoveryState::Completed)
&& matches!(record.state, ReplacementRecoveryState::CleanupPending) =>
{
entry.insert(record);
}
std::collections::btree_map::Entry::Occupied(mut entry) => {
entry.insert(ReplacementRecoveryRecord {
task_id,
state: ReplacementRecoveryState::Unknown,
generation: None,
set_disk_id: None,
target_slots: Vec::new(),
reason: Some("conflicting durable replacement records across survivor disks".to_string()),
verified_at: None,
});
reason.get_or_insert_with(|| "conflicting durable replacement records".to_string());
}
}
}
}
Err(error) => {
reason.get_or_insert_with(|| format!("failed to read local replacement recovery records: {error}"));
}
}
}
ReplacementRecoverySnapshot {
records: records.into_values().collect(),
definitive: reason.is_none(),
reason,
}
}
fn usize_to_u64_saturated(value: usize) -> u64 {
u64::try_from(value).unwrap_or(u64::MAX)
}
+35
View File
@@ -122,6 +122,34 @@ pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
Ok(stat1.st_dev == stat2.st_dev)
}
/// Return whether `path` is an exact Linux mount point.
///
/// Device numbers alone are insufficient here: bind mounts may deliberately
/// share a device number with their source. Auto-format callers use this to
/// refuse an unmounted mountpoint directory rather than writing into its
/// parent filesystem.
pub fn is_mount_point(path: &Path) -> std::io::Result<bool> {
let metadata = fs::symlink_metadata(path)?;
if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
return Ok(false);
}
let canonical_path = fs::canonicalize(path)?;
let mountinfo = fs::read_to_string("/proc/self/mountinfo")?;
let canonical_path = canonical_path.to_string_lossy().replace(' ', "\\040");
Ok(mountinfo
.lines()
.any(|line| mountinfo_path(line) == Some(canonical_path.as_ref())))
}
fn mountinfo_path(line: &str) -> Option<&str> {
let mut fields = line.split_whitespace();
for _ in 0..4 {
fields.next()?;
}
fields.next()
}
/// Resolve the leaf physical device identities backing a local filesystem path.
///
/// Linux block stacks such as partitions, `dm-*`, or software RAID can all
@@ -415,6 +443,13 @@ mod tests {
assert_eq!(paths, vec!["/data/my disk".to_string()]);
}
#[test]
fn mountinfo_path_extracts_the_mountpoint() {
let line = "42 31 8:1 / /data/replacement rw,relatime - ext4 /dev/sda1 rw";
assert_eq!(mountinfo_path(line), Some("/data/replacement"));
}
#[test]
fn reject_relative_path_for_cross_device_validation() {
let err = ensure_no_sub_mounts("relative/path", &[]).unwrap_err();
+4 -3
View File
@@ -24,14 +24,15 @@ mod unix;
mod windows;
#[cfg(target_os = "linux")]
pub use linux::{check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, same_disk};
pub use linux::{check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, is_mount_point, same_disk};
#[cfg(all(unix, not(target_os = "linux")))]
pub use unix::{check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, same_disk};
pub use unix::{check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, is_mount_point, same_disk};
#[cfg(target_os = "windows")]
pub use windows::{
check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, get_volume_serial_number, same_disk,
check_cross_device_mounts, get_drive_stats, get_info, get_physical_device_ids, get_volume_serial_number, is_mount_point,
same_disk,
};
#[derive(Debug, Default, PartialEq)]
+14
View File
@@ -93,6 +93,20 @@ pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
Ok(stat1.st_dev == stat2.st_dev)
}
/// Return whether `path` is mounted separately from its parent directory.
///
/// This is intentionally conservative on non-Linux Unix platforms. Linux
/// uses mountinfo to recognize bind mounts; elsewhere a shared device number
/// cannot distinguish a bind mount from a normal directory safely.
pub fn is_mount_point(path: &Path) -> std::io::Result<bool> {
let metadata = std::fs::symlink_metadata(path)?;
if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
return Ok(false);
}
let parent = path.parent().ok_or_else(|| Error::other("mount point has no parent"))?;
Ok(rustix::fs::stat(path)?.st_dev != rustix::fs::stat(parent)?.st_dev)
}
pub fn get_physical_device_ids(disk: &str) -> std::io::Result<Vec<String>> {
let stat = rustix::fs::stat(disk)?;
let major = rustix::fs::major(stat.st_dev);
+6
View File
@@ -158,6 +158,12 @@ pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
Ok(volume1 == volume2)
}
/// Automatic replacement formatting is disabled until Windows has an
/// equivalent mount identity probe.
pub fn is_mount_point(_path: &std::path::Path) -> std::io::Result<bool> {
Ok(false)
}
pub fn get_physical_device_ids(disk: &str) -> std::io::Result<Vec<String>> {
let path_wide = to_wide_path(Path::new(disk));
let volume = get_volume_name(&path_wide)?;
@@ -213,6 +213,16 @@ missed work. Scanner-originated object checks should appear under
under `admin/high`. If scanner work grows but admin work remains blocked, treat
that as heal queue pressure rather than scanner pacing pressure.
## Replacement Recovery Completion
`POST /v3/background-heal/status` is an execution-queue view. `state=idle`, zero queue and active counts, an online disk, a readable object, or acceptance of an Admin deep-heal request do not independently prove that a replacement disk contains every erasure shard.
Treat replacement recovery as verified only after the repair task has completed for the exact replacement instance and an operator has confirmed the target disk contains the expected `xl.meta` and data parts for every relevant object version. A replacement that is not mounted, is unsafe to format, loses its marker, or returns a partial target outcome must be treated as deferred or incomplete rather than complete.
The v3 route and its peer status protocol preserve their existing fields for mixed-version clusters. A new node must not infer replacement completion from an old or unavailable peer; regard that information as unknown or degraded until every required peer can report the same replacement instance and verified completion. Do not automate destructive replacement actions from an `idle` observation alone.
Replacement resume and checkpoint files use an independent on-disk schema. A newer reader rejects a future schema rather than continuing with data it cannot interpret, while an older binary cannot safely enforce the new generation fence because it may ignore fields it does not know. Do not roll a cluster back after a replacement generation has started. Complete that recovery with the current-or-newer release; if it cannot complete, keep that version for diagnosis rather than deleting its durable records or continuing with an older binary.
## Reading Replication Repair
`metrics.replication_repair`, `metrics.current_cycle_replication_repair`, and