From 47c1e730c70503101ead526f5b175bc4f5297488 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 01:56:03 +0800 Subject: [PATCH] fix(ecstore): make erasure heal write quorum best-effort per target (#4545) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/ecstore/src/erasure/coding/heal.rs | 177 +++++++++++++++++++++- 1 file changed, 175 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/heal.rs b/crates/ecstore/src/erasure/coding/heal.rs index 0d3b1389f..7ab16eb02 100644 --- a/crates/ecstore/src/erasure/coding/heal.rs +++ b/crates/ecstore/src/erasure/coding/heal.rs @@ -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> { + 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> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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::>(); + 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::>(); + + // 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::>(); + + 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::>(); + 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::>(); + + 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::>(); + + erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect_err("heal must fail when all replacement disks fail"); + } }