mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
fix(ecstore): make erasure heal write quorum best-effort per target (#4545)
Erasure heal set write_quorum to the count of available target writers, so every replacement disk had to succeed writing every block or the whole object heal aborted. A single flapping replacement disk failing one block made MultiWriter::write return Err, heal propagate Err, and the ops layer delete the entire tmp staging dir — leaving the other healthy replacement disks unhealed and blocking redundancy recovery indefinitely. Set the heal write_quorum to 1, matching upstream MinIO's writeQuorum=1 for heal. MultiWriter already marks a failed writer as None, and the ops layer (set_disk/ops/heal.rs) already drops the failed writer while committing the survivors; the read-side quorum check and parity cross-checks that guarantee reconstruction correctness are unchanged. Add regression tests: one healthy-and-one-failing target still heals the healthy targets and drops the failing one; all-targets-failing still returns Err so nothing is falsely committed. Fixes #947 Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -132,8 +132,17 @@ impl super::Erasure {
|
||||
end_block += 1;
|
||||
}
|
||||
|
||||
let available_writers = writers.iter().filter(|w| w.is_some()).count();
|
||||
let write_quorum = available_writers.max(1);
|
||||
// Heal is best-effort per target disk. A single replacement disk failing to
|
||||
// write one block must not abort healing on the other healthy replacement
|
||||
// disks: requiring every target to succeed (write_quorum == available_writers)
|
||||
// let one flapping disk block redundancy recovery for the whole object and
|
||||
// kept the erasure set under-protected indefinitely. Upstream MinIO uses
|
||||
// writeQuorum == 1 for heal so partial progress commits; MultiWriter already
|
||||
// marks a failed writer as None (see encode.rs `write_shard`) and the ops
|
||||
// layer (set_disk/ops/heal.rs) then drops the failed writer while committing
|
||||
// the survivors. Reconstruction correctness is still guaranteed by the
|
||||
// read-side quorum check and parity cross-checks below, which are unaffected.
|
||||
let write_quorum = 1;
|
||||
let mut writers = MultiWriter::new(writers, write_quorum);
|
||||
let read_timeout = get_object_disk_read_timeout();
|
||||
let shard_file_size = self.shard_file_size(total_length as i64) as usize;
|
||||
@@ -200,6 +209,35 @@ mod tests {
|
||||
use crate::erasure::coding::{CustomWriter, Erasure};
|
||||
use rustfs_utils::HashAlgorithm;
|
||||
use std::io::Cursor;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use tokio::io::AsyncWrite;
|
||||
|
||||
/// An `AsyncWrite` that accepts writes until `fail_at` bytes have been
|
||||
/// written, then fails every subsequent write. Used to simulate a
|
||||
/// replacement disk that starts failing on a mid-object block.
|
||||
struct FailAfterBytes {
|
||||
written: usize,
|
||||
fail_at: usize,
|
||||
}
|
||||
|
||||
impl AsyncWrite for FailAfterBytes {
|
||||
fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
|
||||
if self.written >= self.fail_at {
|
||||
return Poll::Ready(Err(io::Error::other("injected heal write failure")));
|
||||
}
|
||||
self.written += buf.len();
|
||||
Poll::Ready(Ok(buf.len()))
|
||||
}
|
||||
|
||||
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
|
||||
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn heal_reconstructs_missing_parity_shard() {
|
||||
@@ -389,4 +427,139 @@ mod tests {
|
||||
.expect("inline writer should retain data");
|
||||
assert!(written.is_empty(), "heal must fail before writing rebuilt data");
|
||||
}
|
||||
|
||||
// Regression for backlog#947 (ECA-06): a single replacement disk failing to
|
||||
// write a mid-object block must not abort healing on the other healthy
|
||||
// replacement disks. Before the fix, write_quorum == available_writers made
|
||||
// MultiWriter::write return Err on the first per-writer failure, so heal
|
||||
// aborted the whole object and left the healthy targets unhealed.
|
||||
#[tokio::test]
|
||||
async fn heal_completes_healthy_targets_when_one_target_disk_fails_midway() {
|
||||
// 2 data + 3 parity: two data shards are readable sources, the three
|
||||
// parity positions are empty replacement disks to be healed.
|
||||
let erasure = Erasure::new(2, 3, 64);
|
||||
// Multiple blocks so the injected failure lands on a mid-object block.
|
||||
let data = (0..erasure.block_size * 3)
|
||||
.map(|index| (index % 251) as u8)
|
||||
.collect::<Vec<_>>();
|
||||
let encoded = erasure.encode_data(&data).expect("encode should succeed");
|
||||
|
||||
// Only the two data shards survive as readable sources.
|
||||
let readers = encoded
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(index, shard)| {
|
||||
if index < erasure.data_shards {
|
||||
Some(BitrotReader::new(
|
||||
Cursor::new(shard.to_vec()),
|
||||
erasure.shard_size(),
|
||||
HashAlgorithm::None,
|
||||
false,
|
||||
))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
// Parity index that will fail mid-way; the other two parity targets stay healthy.
|
||||
let failing_target = erasure.data_shards + 1;
|
||||
let mut writers = (0..erasure.total_shard_count())
|
||||
.map(|index| {
|
||||
if index < erasure.data_shards {
|
||||
// Source positions get no writer.
|
||||
None
|
||||
} else if index == failing_target {
|
||||
// Fail after the first block's shard has been written.
|
||||
Some(BitrotWriterWrapper::new(
|
||||
CustomWriter::new_tokio_writer(FailAfterBytes {
|
||||
written: 0,
|
||||
fail_at: erasure.shard_size(),
|
||||
}),
|
||||
erasure.shard_size(),
|
||||
HashAlgorithm::None,
|
||||
))
|
||||
} else {
|
||||
Some(BitrotWriterWrapper::new(
|
||||
CustomWriter::new_inline_buffer(),
|
||||
erasure.shard_size(),
|
||||
HashAlgorithm::None,
|
||||
))
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
erasure
|
||||
.heal(&mut writers, readers, data.len(), &[])
|
||||
.await
|
||||
.expect("heal must succeed on the healthy targets despite one failing disk");
|
||||
|
||||
// The failing target was dropped (marked None) so the ops layer skips it.
|
||||
assert!(writers[failing_target].is_none(), "failing target writer must be dropped, not committed");
|
||||
|
||||
// Every healthy replacement target is fully healed.
|
||||
for index in erasure.data_shards..erasure.total_shard_count() {
|
||||
if index == failing_target {
|
||||
continue;
|
||||
}
|
||||
let healed = writers[index]
|
||||
.take()
|
||||
.expect("healthy target writer should remain")
|
||||
.into_inline_data()
|
||||
.expect("inline writer should retain data");
|
||||
assert_eq!(healed, encoded[index].to_vec(), "healthy target {index} must be fully healed");
|
||||
}
|
||||
}
|
||||
|
||||
// Regression for backlog#947: when every replacement disk fails to write,
|
||||
// heal must return Err (nothing is falsely reported as healed) so the ops
|
||||
// layer cleans up the tmp staging directory instead of committing.
|
||||
#[tokio::test]
|
||||
async fn heal_fails_when_all_target_disks_fail_midway() {
|
||||
let erasure = Erasure::new(2, 3, 64);
|
||||
let data = (0..erasure.block_size * 3)
|
||||
.map(|index| (index % 251) as u8)
|
||||
.collect::<Vec<_>>();
|
||||
let encoded = erasure.encode_data(&data).expect("encode should succeed");
|
||||
|
||||
let readers = encoded
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(index, shard)| {
|
||||
if index < erasure.data_shards {
|
||||
Some(BitrotReader::new(
|
||||
Cursor::new(shard.to_vec()),
|
||||
erasure.shard_size(),
|
||||
HashAlgorithm::None,
|
||||
false,
|
||||
))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let mut writers = (0..erasure.total_shard_count())
|
||||
.map(|index| {
|
||||
if index < erasure.data_shards {
|
||||
None
|
||||
} else {
|
||||
// Every replacement target fails after the first block.
|
||||
Some(BitrotWriterWrapper::new(
|
||||
CustomWriter::new_tokio_writer(FailAfterBytes {
|
||||
written: 0,
|
||||
fail_at: erasure.shard_size(),
|
||||
}),
|
||||
erasure.shard_size(),
|
||||
HashAlgorithm::None,
|
||||
))
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
erasure
|
||||
.heal(&mut writers, readers, data.len(), &[])
|
||||
.await
|
||||
.expect_err("heal must fail when all replacement disks fail");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user