gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-12 23:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(storage): preserve heal failures and timeout responses (#8462)

Browse Source 
* fix(storage): preserve heal failures and timeout responses

Stop repairs with no writable target before opening source readers. Consolidate target failure aggregation, preserve typed causes, and distinguish all-ignored diagnostics from actual successful nil entries. Map typed disk timeouts to retryable S3 responses without changing quorum decisions.

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(heal): unify staging failure cleanup

Route data-stage failures through one local result boundary after temporary
readers and writers are released. Preserve typed report failures, outer
reconstruction errors, partial repair, and the existing commit fences.

Exercise actual source loss after second-part writer admission using the
existing test pause registry. Assert all staged parts are removed, committed
metadata stays unchanged, and restoring the source allows a full-byte retry.

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

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: Chris <anzhengchao@gmail.com>
This commit is contained in:
Hauser
2026-10-11 12:41:13 +08:00
committed by GitHub
parent 93a64fdca3
commit 0263d5fcb9
5 changed files with 555 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/error_reduce.rs
+27 -4
View File
@@ -131,6 +131,15 @@ pub fn build_write_quorum_failure_summary(
fn dominant_error_label(errors: &[Option<Error>], ignored_errs: &[Error], dominant_error: Option<&Error>) -> &'static str {
let Some(dominant_error) = dominant_error else {
// The reducer also returns None when every failure is ignored for
// quorum voting. That case contains no successful nil entry.
if !errors.is_empty() && errors.iter().all(Option::is_some) {
return if errors.iter().all(|error| error.as_ref() == Some(&Error::DiskNotFound)) {
"disk_not_found"
} else {
"all_ignored"
};
}
return "nil_dominated";
};
@@ -387,9 +396,9 @@ mod tests {
}
#[test]
fn test_build_write_quorum_failure_summary_all_ignored_is_nil_dominated() {
// All disks offline (all DiskNotFound). The dominant error must be None and
// the label must be "nil_dominated" rather than falling back to "other_error".
fn test_build_write_quorum_failure_summary_all_missing_is_disk_not_found() {
// Ignoring unavailable disks for quorum reduction must not imply a
// successful nil entry in the diagnostic label.
let errors = vec![
Some(Error::DiskNotFound),
Some(Error::DiskNotFound),
@@ -397,7 +406,21 @@ mod tests {
];
let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2);
assert_eq!(summary.dominant_error, None);
assert_eq!(summary.dominant_error_label, "nil_dominated");
assert_eq!(summary.achieved, 0);
assert_eq!(summary.dominant_error_label, "disk_not_found");
}
#[test]
fn test_build_write_quorum_failure_summary_all_ignored_has_no_success() {
let errors = [Some(Error::FaultyDisk), Some(Error::DiskNotFound)];
let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2);
assert_eq!(summary.dominant_error, None);
assert_eq!(summary.achieved, 0);
assert_eq!(summary.dominant_error_label, "all_ignored");
assert_eq!(
reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2),
Some(Error::ErasureWriteQuorum)
);
}
#[test]
crates/ecstore/src/set_disk/ops/heal.rs
+489 -304
View File
@@ -99,6 +99,7 @@ static HEAL_RENAME_FAILURES: std::sync::Mutex<Vec<(String, String, usize)>> = st
#[derive(Clone, Copy, PartialEq, Eq)]
enum ReadRepairPausePhase {
Data,
PartStaging(usize),
Commit,
}
@@ -250,7 +251,16 @@ fn should_fail_heal_rename(_bucket: &str, _object: &str, _disk_index: usize) ->
}
#[cfg(test)]
static HEAL_WRITER_FAILURES: std::sync::Mutex<Vec<(String, String, usize, DiskError)>> = std::sync::Mutex::new(Vec::new());
struct HealWriterFailure {
bucket: String,
object: String,
part_number: Option<usize>,
disk_index: usize,
error: DiskError,
}
#[cfg(test)]
static HEAL_WRITER_FAILURES: std::sync::Mutex<Vec<HealWriterFailure>> = std::sync::Mutex::new(Vec::new());
#[cfg(test)]
struct HealWriterFailureScope {
@@ -260,21 +270,23 @@ struct HealWriterFailureScope {
#[cfg(test)]
impl HealWriterFailureScope {
fn install(bucket: &str, object: &str, disk_indexes: &[usize], error: DiskError) -> Self {
fn install(bucket: &str, object: &str, part_number: Option<usize>, disk_indexes: &[usize], error: DiskError) -> Self {
let mut failures = HEAL_WRITER_FAILURES
.lock()
.expect("heal writer failure registry should not poison");
assert!(
!failures.iter().any(|(registered_bucket, registered_object, _, _)| {
registered_bucket == bucket && registered_object == object
}),
!failures
.iter()
.any(|failure| failure.bucket == bucket && failure.object == object),
"heal writer failures must be installed once per object"
);
failures.extend(
disk_indexes
.iter()
.map(|index| (bucket.to_string(), object.to_string(), *index, error.clone())),
);
failures.extend(disk_indexes.iter().map(|index| HealWriterFailure {
bucket: bucket.to_string(),
object: object.to_string(),
part_number,
disk_index: *index,
error: error.clone(),
}));
Self {
bucket: bucket.to_string(),
object: object.to_string(),
@@ -288,25 +300,28 @@ impl Drop for HealWriterFailureScope {
HEAL_WRITER_FAILURES
.lock()
.expect("heal writer failure registry should not poison")
.retain(|(bucket, object, _, _)| bucket != &self.bucket || object != &self.object);
.retain(|failure| failure.bucket != self.bucket || failure.object != self.object);
}
}
#[cfg(test)]
fn injected_heal_writer_error(bucket: &str, object: &str, disk_index: usize) -> Option<DiskError> {
fn injected_heal_writer_error(bucket: &str, object: &str, part_number: usize, disk_index: usize) -> Option<DiskError> {
let mut failures = HEAL_WRITER_FAILURES
.lock()
.expect("heal writer failure registry should not poison");
failures
.iter()
.position(|(registered_bucket, registered_object, registered_index, _)| {
registered_bucket == bucket && registered_object == object && *registered_index == disk_index
.position(|failure| {
failure.bucket == bucket
&& failure.object == object
&& failure.disk_index == disk_index
&& failure.part_number.is_none_or(|part| part == part_number)
})
.map(|position| failures.swap_remove(position).3)
.map(|position| failures.swap_remove(position).error)
}
#[cfg(not(test))]
fn injected_heal_writer_error(_bucket: &str, _object: &str, _disk_index: usize) -> Option<DiskError> {
fn injected_heal_writer_error(_bucket: &str, _object: &str, _part_number: usize, _disk_index: usize) -> Option<DiskError> {
None
}
@@ -518,30 +533,64 @@ fn is_retryable_heal_writer_error(error: &DiskError) -> bool {
) || error.is_retryable_internode_write_failure()
}
fn warn_heal_writer_failures(
bucket: &str,
object: &str,
version_id: &str,
writer_failure_count: usize,
result: &'static str,
first_failure: &(usize, usize, String),
) {
let (first_part_number, first_disk_index, first_error) = first_failure;
warn!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket,
object,
version_id,
writer_failure_count,
first_part_number,
first_disk_index,
error = %first_error,
result,
state = "writer_unavailable",
"Set disk object heal writer failures"
);
struct HealWriterFailures {
target_count: usize,
count: usize,
first: Option<(usize, usize, String)>,
error: Option<DiskError>,
}
impl HealWriterFailures {
fn new(target_count: usize) -> Self {
Self {
target_count,
count: 0,
first: None,
error: None,
}
}
fn record(&mut self, part_number: usize, disk_index: usize, error: DiskError) {
self.count += 1;
if self.first.is_none() {
self.first = Some((part_number, disk_index, heal_writer_error_summary(&error)));
}
// A permanent failure must not be hidden behind an earlier transient
// target failure when the remaining targets also become unavailable.
if self
.error
.as_ref()
.is_none_or(|first| is_retryable_heal_writer_error(first) && !is_retryable_heal_writer_error(&error))
{
self.error = Some(error);
}
}
fn into_error(self) -> DiskError {
self.error.unwrap_or(DiskError::ErasureWriteQuorum)
}
fn warn(&self, bucket: &str, object: &str, version_id: &str, result: &'static str) {
let Some((first_part_number, first_disk_index, first_error)) = &self.first else {
return;
};
warn!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "writer_unavailable",
result,
bucket,
object,
version_id,
target_count = self.target_count,
writer_failure_count = self.count,
first_part_number,
first_disk_index,
error = %first_error,
"Set disk object heal writer failures"
);
}
}
impl SetDisks {
@@ -1548,196 +1597,169 @@ impl SetDisks {
let dst_data_dir = data_dir;
if !latest_meta.deleted && !latest_meta.is_remote() {
let erasure_info = latest_meta.erasure.clone();
let mut writer_failure_count = 0usize;
let mut first_writer_failure = None;
let mut first_writer_error = None;
let mut all_writer_failures_retryable = true;
let mut writer_failure_warned = false;
let stage_result: disk::error::Result<Option<DiskError>> = async {
let erasure_info = latest_meta.erasure.clone();
let mut writer_failures = HealWriterFailures::new(disks_to_heal_count);
for (part_index, part) in latest_meta.parts.iter().enumerate() {
use crate::io_support::shard_integrity::{PartProofReader, PreparedIntegrity, ShardVerifier};
let proof = part
.integrity
.as_ref()
.map(|part| {
PartProofReader::new(part.clone(), &parts_metadata, &latest_disks, bucket, object)
})
.transpose()
.map_err(DiskError::from)?;
// This reader covers the whole encoded shard. A range
// offset can exclude the final encoded byte when the
// logical part ends inside a data block, allowing a
// one-byte tail truncation to evade deep Heal.
let till_offset = usize::try_from(erasure.shard_file_size(part.size as i64))
.map_err(|_| DiskError::FileCorrupt)?;
let use_mmap_read = object_mmap_read_enabled();
for (part_index, part) in latest_meta.parts.iter().enumerate() {
use crate::io_support::shard_integrity::{PartProofReader, PreparedIntegrity, ShardVerifier};
let proof = part
.integrity
.as_ref()
.map(|part| {
PartProofReader::new(part.clone(), &parts_metadata, &latest_disks, bucket, object)
})
.transpose()?;
// Cover the complete encoded shard, including a logical
// part's partial final data block, when checking truncation.
let till_offset = usize::try_from(erasure.shard_file_size(part.size as i64))
.map_err(|_| DiskError::FileCorrupt)?;
let use_mmap_read = object_mmap_read_enabled();
let mut readers = Vec::with_capacity(latest_disks.len());
let mut writers = Vec::with_capacity(out_dated_disks.len());
// let mut errors = Vec::with_capacity(out_dated_disks.len());
let mut writers = Vec::with_capacity(out_dated_disks.len());
let mut prefer = vec![false; latest_disks.len()];
for (index, disk) in latest_disks.iter().enumerate() {
// Preserve the committed layout: recomputing inline-ness here
// (with a hardcoded unversioned threshold) makes healed replicas
// diverge from healthy ones in quorum identity, so heal would
// flag them forever.
let is_inline_buffer = latest_meta.inline_data();
// create writers for all disk positions, but only for outdated disks
for (index, disk_op) in out_dated_disks.iter().enumerate() {
if let Some(outdated_disk) = disk_op {
let writer_result = if let Some(error) =
injected_heal_writer_error(bucket, object, part.number, index)
{
Err(error)
} else {
create_bitrot_writer(
is_inline_buffer,
Some(outdated_disk),
RUSTFS_META_TMP_BUCKET,
&path_join_buf(&[
&tmp_id.to_string(),
&dst_data_dir.to_string(),
&format!("part.{}", part.number),
]),
erasure.shard_file_size(part.size as i64),
erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
)
.await
};
let writer = match writer_result {
Ok(writer) => writer,
Err(err) => {
writer_failures.record(part.number, index, err);
writers.push(None);
continue;
}
};
let mut writer = writer;
if let Some(proof) = &proof {
let verifier = ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)?;
writer.set_integrity(verifier);
}
writers.push(Some(writer));
} else {
writers.push(None);
}
}
// Heal each part. erasure.Heal() will write the healed
// part to .rustfs/tmp/uuid/ which needs to be renamed
// later to the final location.
if writers.iter().all(Option::is_none) {
writer_failures.warn(bucket, object, version_id, "all_targets_unavailable");
// No target can accept this part. Preserve the setup
// cause before opening source readers or reconstructing.
return Ok(Some(writer_failures.into_error()));
}
#[cfg(test)]
pause_read_repair(bucket, object, ReadRepairPausePhase::PartStaging(part.number)).await;
let mut readers = Vec::with_capacity(latest_disks.len());
let mut prefer = vec![false; latest_disks.len()];
let this_part_errs =
Self::shuffle_check_parts(&data_errs_by_part[&part_index], &erasure_info.distribution);
if this_part_errs[index] != CHECK_PART_SUCCESS {
trace!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
part_number = part.number,
disk_index = index,
part_status = this_part_errs[index],
state = "source_shard_skipped",
"Set disk source shard skipped"
);
readers.push(None);
continue;
}
if let (Some(disk), Some(metadata)) = (disk, &copy_parts_metadata[index]) {
let checksum_info = metadata.erasure.get_checksum_info(part.number);
let checksum_algo = if metadata.uses_legacy_checksum
&& checksum_info.algorithm == HashAlgorithm::HighwayHash256S
{
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
match create_bitrot_reader(
metadata.data.as_deref(),
Some(disk),
bucket,
&path_join_buf(&[object, &src_data_dir, &format!("part.{}", part.number)]),
0,
till_offset,
erasure.shard_size(),
checksum_algo.clone(),
false,
use_mmap_read,
)
.await
{
Ok(Some(mut reader)) => {
if let Some(proof) = &proof {
reader
.set_integrity(
ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)
.map_err(DiskError::from)?,
)
.map_err(DiskError::from)?;
}
readers.push(Some(reader));
}
Ok(None) => {
readers.push(None);
continue;
}
Err(_e) => {
readers.push(None);
continue;
}
for (index, disk) in latest_disks.iter().enumerate() {
if this_part_errs[index] != CHECK_PART_SUCCESS {
trace!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
part_number = part.number,
disk_index = index,
part_status = this_part_errs[index],
state = "source_shard_skipped",
"Set disk source shard skipped"
);
readers.push(None);
continue;
}
prefer[index] = disk.host_name().is_empty();
} else {
readers.push(None);
// errors.push(Some(DiskError::DiskNotFound));
}
}
if let (Some(disk), Some(metadata)) = (disk, &copy_parts_metadata[index]) {
let checksum_info = metadata.erasure.get_checksum_info(part.number);
let checksum_algo = if metadata.uses_legacy_checksum
&& checksum_info.algorithm == HashAlgorithm::HighwayHash256S
{
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
// Preserve the committed layout: recomputing inline-ness here
// (with a hardcoded unversioned threshold) makes healed replicas
// diverge from healthy ones in quorum identity, so heal would
// flag them forever.
let is_inline_buffer = latest_meta.inline_data();
// create writers for all disk positions, but only for outdated disks
for (index, disk_op) in out_dated_disks.iter().enumerate() {
if let Some(outdated_disk) = disk_op {
let writer_result = if let Some(error) = injected_heal_writer_error(bucket, object, index)
{
Err(error)
} else {
create_bitrot_writer(
is_inline_buffer,
Some(outdated_disk),
RUSTFS_META_TMP_BUCKET,
&path_join_buf(&[
&tmp_id.to_string(),
&dst_data_dir.to_string(),
&format!("part.{}", part.number),
]),
erasure.shard_file_size(part.size as i64),
match create_bitrot_reader(
metadata.data.as_deref(),
Some(disk),
bucket,
&path_join_buf(&[object, &src_data_dir, &format!("part.{}", part.number)]),
0,
till_offset,
erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
checksum_algo.clone(),
false,
use_mmap_read,
)
.await
};
let writer = match writer_result {
Ok(writer) => writer,
Err(err) => {
writer_failure_count += 1;
all_writer_failures_retryable &= is_retryable_heal_writer_error(&err);
if first_writer_failure.is_none() {
first_writer_failure =
Some((part.number, index, heal_writer_error_summary(&err)));
first_writer_error = Some(err);
{
Ok(Some(mut reader)) => {
if let Some(proof) = &proof {
let verifier =
ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)?;
reader.set_integrity(verifier)?;
}
readers.push(Some(reader));
}
Ok(None) => {
readers.push(None);
continue;
}
Err(_e) => {
readers.push(None);
continue;
}
writers.push(None);
continue;
}
};
let mut writer = writer;
if let Some(proof) = &proof {
writer.set_integrity(
ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)
.map_err(DiskError::from)?,
);
}
writers.push(Some(writer));
} else {
writers.push(None);
}
}
// Heal each part. erasure.Heal() will write the healed
// part to .rustfs/tmp/uuid/ which needs to be renamed
// later to the final location.
if writer_failure_count > 0
&& writers.iter().all(Option::is_none)
&& let Some(first_failure) = first_writer_failure.as_ref()
{
warn_heal_writer_failures(
bucket,
object,
version_id,
writer_failure_count,
"all_targets_unavailable",
first_failure,
);
writer_failure_warned = true;
}
let heal_part = erasure.heal(&mut writers, readers, part.size, &prefer);
let heal_part_result = if read_repair_uses_shared_lock {
match tokio::time::timeout(READ_REPAIR_DATA_PHASE_TIMEOUT, heal_part).await {
Ok(result) => result,
Err(_) => Err(DiskError::Timeout),
prefer[index] = disk.host_name().is_empty();
} else {
readers.push(None);
// errors.push(Some(DiskError::DiskNotFound));
}
}
} else {
heal_part.await
};
if let Err(e) = heal_part_result {
// Don't leak the partially-written healed shards in
// .rustfs/tmp when heal fails midway (backlog#799 B20).
let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
return Err(e);
}
// close_bitrot_writers(&mut writers).await?;
if !is_inline_buffer && let Some(proof) = &proof {
let before = out_dated_disks.iter().filter(|disk| disk.is_some()).count();
let proof_result = async {
let heal_part = erasure.heal(&mut writers, readers, part.size, &prefer);
let heal_part_result = if read_repair_uses_shared_lock {
match tokio::time::timeout(READ_REPAIR_DATA_PHASE_TIMEOUT, heal_part).await {
Ok(result) => result,
Err(_) => Err(DiskError::Timeout),
}
} else {
heal_part.await
};
heal_part_result?;
// close_bitrot_writers(&mut writers).await?;
if !is_inline_buffer && let Some(proof) = &proof {
let before = out_dated_disks.iter().filter(|disk| disk.is_some()).count();
PreparedIntegrity::copy_from(proof)
.await?
.write(
@@ -1746,89 +1768,71 @@ impl SetDisks {
RUSTFS_META_TMP_BUCKET,
&format!("{tmp_id}/{dst_data_dir}"),
)
.await
}
.await;
if let Err(error) = proof_result {
let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
return Err(error.into());
}
disks_to_heal_count -= before - out_dated_disks.iter().filter(|disk| disk.is_some()).count();
}
for (index, disk_op) in out_dated_disks.iter_mut().enumerate() {
if disk_op.is_none() {
continue;
.await?;
disks_to_heal_count -=
before - out_dated_disks.iter().filter(|disk| disk.is_some()).count();
}
if writers[index].is_none() {
*disk_op = None;
disks_to_heal_count -= 1;
continue;
}
parts_metadata[index].data_dir = Some(dst_data_dir);
parts_metadata[index].add_object_part(
part.number,
part.etag.clone(),
part.size,
part.mod_time,
part.actual_size,
part.index.clone(),
part.checksums.clone(),
);
if let Some(updated) = parts_metadata[index]
.parts
.iter_mut()
.find(|updated| updated.number == part.number)
{
updated.integrity.clone_from(&part.integrity);
}
if is_inline_buffer {
if let Some(writer) = writers[index].take() {
// if let Some(w) = writer.as_any().downcast_ref::<BitrotFileWriter>() {
// parts_metadata[index].data = Some(w.inline_data().to_vec());
// }
parts_metadata[index].data =
Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
for (index, disk_op) in out_dated_disks.iter_mut().enumerate() {
if disk_op.is_none() {
continue;
}
parts_metadata[index].set_inline_data();
} else {
parts_metadata[index].data = None;
}
}
if disks_to_heal_count == 0 {
if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() {
warn_heal_writer_failures(
bucket,
object,
version_id,
writer_failure_count,
"all_targets_unavailable",
first_failure,
if writers[index].is_none() {
*disk_op = None;
disks_to_heal_count -= 1;
continue;
}
parts_metadata[index].data_dir = Some(dst_data_dir);
parts_metadata[index].add_object_part(
part.number,
part.etag.clone(),
part.size,
part.mod_time,
part.actual_size,
part.index.clone(),
part.checksums.clone(),
);
if let Some(updated) = parts_metadata[index]
.parts
.iter_mut()
.find(|updated| updated.number == part.number)
{
updated.integrity.clone_from(&part.integrity);
}
if is_inline_buffer {
if let Some(writer) = writers[index].take() {
// if let Some(w) = writer.as_any().downcast_ref::<BitrotFileWriter>() {
// parts_metadata[index].data = Some(w.inline_data().to_vec());
// }
parts_metadata[index].data =
Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
}
parts_metadata[index].set_inline_data();
} else {
parts_metadata[index].data = None;
}
}
// Clean up healed shards written to .rustfs/tmp before bailing (B20).
let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
let error = if all_writer_failures_retryable {
first_writer_error.take().unwrap_or(DiskError::FaultyDisk)
} else {
DiskError::other(format!("all drives had write errors, unable to heal {bucket}/{object}"))
};
return Ok((result, Some(error)));
}
}
if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() {
warn_heal_writer_failures(
bucket,
object,
version_id,
writer_failure_count,
"partial_targets_unavailable",
first_failure,
);
if disks_to_heal_count == 0 {
writer_failures.warn(bucket, object, version_id, "all_targets_unavailable");
return Ok(Some(writer_failures.into_error()));
}
}
writer_failures.warn(bucket, object, version_id, "partial_targets_unavailable");
Ok(None)
}
.await;
match stage_result {
Ok(None) => {}
failure => {
// Release readers and writers before removing all parts
// staged under this heal's exclusively owned prefix.
let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
return failure.map(|error| (result, error));
}
}
}
let _read_repair_write_lock_guard = if let Some(expected_fingerprint) = read_repair_commit_fingerprint {
@@ -3531,8 +3535,8 @@ impl SetDisks {
#[cfg(test)]
mod heal_result_report_tests {
use super::{
AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, ReadRepairCommitFingerprint,
ReadRepairPauseScope, SetDisks, heal_writer_error_summary, is_retryable_heal_writer_error,
AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, HealWriterFailures,
ReadRepairCommitFingerprint, ReadRepairPauseScope, SetDisks, heal_writer_error_summary, is_retryable_heal_writer_error,
};
use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope, HealWriterFailureScope};
use crate::disk::endpoint::Endpoint;
@@ -3786,10 +3790,17 @@ mod heal_result_report_tests {
#[tokio::test(flavor = "current_thread")]
#[serial_test::serial]
async fn heal_writer_failures_emit_one_aggregate_warning_per_object() {
for (case, failed_target_count, expected_result, expect_error) in [
("partial", 1usize, "partial_targets_unavailable", false),
("all", 2usize, "all_targets_unavailable", true),
for (case, target_count, failed_target_count, injected_error) in [
("partial", 2usize, 1usize, DiskError::DiskFull),
("all-full", 2, 2, DiskError::DiskFull),
("all-timeout", 2, 2, DiskError::Timeout),
("single-faulty", 1, 1, DiskError::FaultyDisk),
] {
let expected_result = if target_count == failed_target_count {
"all_targets_unavailable"
} else {
"partial_targets_unavailable"
};
let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = format!("heal-writer-{case}");
let object = "object.bin";
@@ -3816,11 +3827,13 @@ mod heal_result_report_tests {
.read_version("", &bucket, object, "", &ReadOptions::default())
.await
.expect("source metadata should be readable");
let mut target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1];
let mut target_slots = (0..target_count)
.map(|index| source.erasure.distribution[index] - 1)
.collect::<Vec<_>>();
target_slots.sort_unstable();
for index in [0, 1] {
remove_current_object_part(&temp_dirs[index], &bucket, object)
for dir in temp_dirs.iter().take(target_count) {
remove_current_object_part(dir, &bucket, object)
.await
.expect("target shard should be removed before heal");
}
@@ -3834,7 +3847,9 @@ mod heal_result_report_tests {
.without_time()
.finish();
let subscriber_guard = tracing::subscriber::set_default(subscriber);
let failure_scope = HealWriterFailureScope::install(&bucket, object, failed_slots, DiskError::DiskFull);
let tmp_before = non_trash_tmp_entries(&temp_dirs).await;
let injected_summary = heal_writer_error_summary(&injected_error);
let failure_scope = HealWriterFailureScope::install(&bucket, object, None, failed_slots, injected_error.clone());
let heal_outcome = set
.heal_object(
@@ -3851,12 +3866,22 @@ mod heal_result_report_tests {
drop(failure_scope);
drop(subscriber_guard);
let (_, error) = heal_outcome.expect("writer setup failures must retain a heal report and their typed error");
assert_eq!(
heal_outcome.is_err(),
expect_error,
"{case}: aggregate heal result should match writer outcomes"
error,
(target_count == failed_target_count).then_some(injected_error),
"{case}: partial repair succeeds and all-target failure retains its cause"
);
assert_eq!(
non_trash_tmp_entries(&temp_dirs).await,
tmp_before,
"{case}: heal must clean temporary shards"
);
let output = logs.contents();
assert!(
!output.contains("Erasure encode write quorum unavailable"),
"{case}: unavailable heal targets must not fabricate an all-disk encode failure: {output}"
);
assert_eq!(
output.matches("Set disk object heal writer failures").count(),
1,
@@ -3879,12 +3904,172 @@ mod heal_result_report_tests {
"{case}: warning must distinguish partial from all-target failure: {output}"
);
assert!(
output.contains("error=drive path full"),
output.contains(&format!("error={injected_summary}")),
"{case}: warning must preserve a redacted failure reason: {output}"
);
}
}
#[test]
fn heal_writer_failure_aggregation_retains_permanent_cause() {
for errors in [
[DiskError::Timeout, DiskError::DiskFull],
[DiskError::DiskFull, DiskError::Timeout],
] {
let mut failures = HealWriterFailures::new(2);
for (index, error) in errors.into_iter().enumerate() {
failures.record(1, index, error);
}
assert_eq!(failures.count, 2);
assert_eq!(failures.into_error(), DiskError::DiskFull);
}
}
#[tokio::test]
#[serial_test::serial]
async fn heal_second_part_failures_clean_staged_parts_and_retry() {
use crate::set_disk::CompletePart;
use crate::storage_api_contracts::multipart::MultipartOperations as _;
use tokio::io::AsyncReadExt as _;
let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = "heal-writer-second-part";
let object = "multipart.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("fixture bucket");
}
let options = ObjectOptions {
no_lock: true,
..Default::default()
};
let upload = set
.new_multipart_upload(bucket, object, &options)
.await
.expect("multipart upload");
let payloads = [vec![0x35; 5 * 1024 * 1024], vec![0x69; 1024 * 1024]];
let mut completed = Vec::new();
for (index, payload) in payloads.iter().enumerate() {
let part = set
.put_object_part(
bucket,
object,
&upload.upload_id,
index + 1,
&mut PutObjReader::from_vec(payload.clone()),
&options,
)
.await
.expect("multipart part");
completed.push(CompletePart {
part_num: part.part_num,
etag: part.etag,
..Default::default()
});
}
set.clone()
.complete_multipart_upload(bucket, object, &upload.upload_id, completed, &options)
.await
.expect("completed multipart object");
let source = disks[2]
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("healthy source metadata");
let data_dir = source.data_dir.expect("multipart data directory").to_string();
let target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1];
for dir in &dirs[..2] {
tokio::fs::remove_dir_all(dir.path().join(bucket).join(object).join(&data_dir))
.await
.expect("remove target shards");
}
let tmp_before = non_trash_tmp_entries(&dirs).await;
let scope = HealWriterFailureScope::install(bucket, object, Some(2), &target_slots, DiskError::Timeout);
let heal_options = HealOpts {
no_lock: true,
scan_mode: HealScanMode::Deep,
..Default::default()
};
let (_, error) = set
.heal_object(bucket, object, "", &heal_options)
.await
.expect("failed heal report");
drop(scope);
assert_eq!(error, Some(DiskError::Timeout));
assert_eq!(
non_trash_tmp_entries(&dirs).await,
tmp_before,
"first-part staging must be cleaned after the second-part setup failure"
);
let source_path = dirs[2].path().join(bucket).join(object).join(&data_dir).join("part.2");
let source_bytes = tokio::fs::read(&source_path).await.expect("committed source shard");
let mut metadata_before = Vec::new();
for dir in &dirs {
metadata_before.push(
tokio::fs::read(dir.path().join(bucket).join(object).join("xl.meta"))
.await
.expect("committed metadata before reconstruction failure"),
);
}
let (pause_scope, arrived, release) =
ReadRepairPauseScope::install_at(bucket, object, super::ReadRepairPausePhase::PartStaging(2));
let lose_source = async {
arrived.notified().await;
assert_ne!(
non_trash_tmp_entries(&dirs).await,
tmp_before,
"the first part must already be staged before the source loss"
);
tokio::fs::remove_file(&source_path)
.await
.expect("lose one remaining source shard");
release.notify_one();
};
let (outcome, ()) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
tokio::join!(set.heal_object(bucket, object, "", &heal_options), lose_source)
})
.await
.expect("heal must report source loss without waiting indefinitely");
drop(pause_scope);
assert_eq!(
outcome.expect_err("source loss after writer admission must preserve the outer failure"),
DiskError::ErasureReadQuorum
);
assert_eq!(
non_trash_tmp_entries(&dirs).await,
tmp_before,
"reconstruction failure must clean every staged part"
);
for (dir, expected) in dirs.iter().zip(metadata_before) {
assert_eq!(
tokio::fs::read(dir.path().join(bucket).join(object).join("xl.meta"))
.await
.expect("metadata after reconstruction failure"),
expected,
"a failed data stage must not publish replacement metadata"
);
}
tokio::fs::write(&source_path, source_bytes)
.await
.expect("restore the committed source shard");
let (_, error) = set
.heal_object(bucket, object, "", &heal_options)
.await
.expect("retry after target recovery");
assert!(error.is_none(), "recovered targets should heal both parts: {error:?}");
let mut reader = set
.get_object_reader(bucket, object, None, Default::default(), &options)
.await
.expect("read repaired multipart object");
let mut actual = Vec::new();
reader
.stream
.read_to_end(&mut actual)
.await
.expect("complete repaired object body");
assert_eq!(actual, payloads.concat());
}
async fn real_disk() -> (TempDir, Endpoint, DiskStore) {
let dir = tempfile::tempdir().expect("tempdir should be created");
let endpoint =
rustfs/src/error.rs
+32
View File
@@ -695,6 +695,7 @@ impl From<StorageError> for ApiError {
StorageError::FaultyDisk
| StorageError::FaultyRemoteDisk
| StorageError::DiskNotFound
| StorageError::Timeout
| StorageError::TooManyOpenFiles => S3ErrorCode::ServiceUnavailable,
StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => {
S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())
@@ -1719,6 +1720,37 @@ mod tests {
}
}
#[test]
fn disk_quorum_timeouts_map_to_retryable_service_unavailable() {
use crate::storage_api::error::{DiskError, OBJECT_OP_IGNORED_ERRS, reduce_read_quorum_errs, reduce_write_quorum_errs};
let errors = [
Some(DiskError::Timeout),
Some(DiskError::Timeout),
Some(DiskError::Timeout),
None,
];
for reduced in [
reduce_read_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2),
reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 3),
] {
let reduced = reduced.expect("quorum of timeouts must retain its typed cause");
assert_eq!(reduced, DiskError::Timeout);
let error = ApiError::from(StorageError::from(reduced));
assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
let response = S3Error::from(error);
assert_eq!(response.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
}
let degraded = [None, None, None, Some(DiskError::Timeout)];
assert!(reduce_write_quorum_errs(&degraded, OBJECT_OP_IGNORED_ERRS, 3).is_none());
for failure in [DiskError::FileCorrupt, DiskError::Io(IoError::other("unknown disk failure"))] {
let errors = [Some(failure.clone()), Some(failure.clone()), Some(failure), None];
let reduced =
reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 3).expect("permanent failures must remain errors");
assert_eq!(ApiError::from(StorageError::from(reduced)).code, S3ErrorCode::InternalError);
}
}
#[test]
fn test_api_error_from_unexpected_eof_maps_to_incomplete_body() {
let io_error = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 7 });
rustfs/src/storage_api.rs
+6
View File
@@ -79,6 +79,12 @@ pub(crate) mod config_test {
}
pub(crate) mod error {
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_disk::error::Error as DiskError;
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_disk::error_reduce::{
OBJECT_OP_IGNORED_ERRS, reduce_read_quorum_errs, reduce_write_quorum_errs,
};
pub(crate) mod contract {
pub(crate) use super::super::storage_contracts::error::StorageErrorCode;
scripts/error-other-format-baseline.txt
+1 -1
View File
@@ -59,7 +59,7 @@
7|crates/ecstore/src/set_disk/core/io_primitives.rs
1|crates/ecstore/src/set_disk/mod.rs
3|crates/ecstore/src/set_disk/ops/bitrot_self_verify.rs
2|crates/ecstore/src/set_disk/ops/heal.rs
1|crates/ecstore/src/set_disk/ops/heal.rs
8|crates/ecstore/src/set_disk/ops/multipart.rs
2|crates/ecstore/src/set_disk/ops/object.rs
3|crates/ecstore/src/set_disk/read.rs