From ea74fa7a959445238955e8ebfff5d434b714f61a Mon Sep 17 00:00:00 2001 From: weisd Date: Tue, 26 May 2026 17:49:25 +0800 Subject: [PATCH] fix(heal): rebuild parity shards during repair (#3086) Object repair rebuilt missing data shards but left missing parity shards unrecreated. A parity-only repair could therefore complete without restoring the shard contents. Add a repair-specific reconstruction path that regenerates parity after data shards are available, while keeping normal read decoding data-only. Also shut down repair writers after all blocks are written. Constraint: Preserve read-path decode behavior and limit parity regeneration to repair. Confidence: high Scope-risk: moderate Directive: Do not route read decoding through parity regeneration without measuring the cost. Tested: cargo test -p rustfs-ecstore decode_data_and_parity -- --nocapture Tested: cargo test -p rustfs-ecstore heal_reconstructs_missing_parity_shard -- --nocapture Tested: cargo test -p rustfs-ecstore erasure_coding -- --nocapture Tested: cargo test -p rustfs heal_object_marks_missing_shard_disk_dirty_for_capacity_manager -- --nocapture Tested: cargo test -p rustfs-heal -- --nocapture Tested: cargo clippy -p rustfs-ecstore --all-targets -- -D warnings Tested: cargo fmt --all --check Tested: make pre-commit Not-tested: Live multi-node disk replacement outside local test harness --- crates/ecstore/src/erasure_coding/erasure.rs | 213 ++++++++++++++++++- crates/ecstore/src/erasure_coding/heal.rs | 125 ++++++++++- 2 files changed, 327 insertions(+), 11 deletions(-) diff --git a/crates/ecstore/src/erasure_coding/erasure.rs b/crates/ecstore/src/erasure_coding/erasure.rs index c83f1e028..58e6229cf 100644 --- a/crates/ecstore/src/erasure_coding/erasure.rs +++ b/crates/ecstore/src/erasure_coding/erasure.rs @@ -106,7 +106,7 @@ impl LegacyReedSolomonEncoder { Ok(()) } - fn reconstruct(&self, shards: &mut [Option>]) -> io::Result<()> { + fn reconstruct_data(&self, shards: &mut [Option>]) -> io::Result<()> { let shard_len = shards .iter() .find_map(|s| s.as_ref().map(|v| v.len())) @@ -172,6 +172,15 @@ impl LegacyReedSolomonEncoder { Ok(()) } + + fn reconstruct(&self, shards: &mut [Option>]) -> io::Result<()> { + self.reconstruct_data(shards)?; + self.encode_parity(shards) + } + + fn encode_parity(&self, shards: &mut [Option>]) -> io::Result<()> { + encode_parity_shards(shards, self.data_shards, self.parity_shards, |shards| self.encode(shards)) + } } /// Reed-Solomon encoder using reed-solomon-erasure @@ -224,8 +233,8 @@ impl ReedSolomonEncoder { } } - /// Reconstruct missing shards. - pub fn reconstruct(&self, shards: &mut [Option>]) -> io::Result<()> { + /// Reconstruct missing data shards. + pub fn reconstruct_data(&self, shards: &mut [Option>]) -> io::Result<()> { if let Some(ref rs) = self.encoder { rs.reconstruct_data(shards) .map_err(|e| io::Error::other(format!("Reed-Solomon reconstruct failed: {e:?}"))) @@ -233,6 +242,58 @@ impl ReedSolomonEncoder { Ok(()) } } + + /// Reconstruct missing data shards and regenerate parity shards. + pub fn reconstruct(&self, shards: &mut [Option>]) -> io::Result<()> { + self.reconstruct_data(shards)?; + self.encode_parity(shards) + } + + fn encode_parity(&self, shards: &mut [Option>]) -> io::Result<()> { + encode_parity_shards(shards, self.data_shards, self.parity_shards, |shards| self.encode(shards)) + } +} + +fn encode_parity_shards(shards: &mut [Option>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()> +where + F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>, +{ + let expected_shards = data_shards + parity_shards; + if shards.len() != expected_shards { + return Err(io::Error::other(format!( + "invalid shard count: got {}, expected {}", + shards.len(), + expected_shards + ))); + } + + let shard_len = shards + .iter() + .find_map(|s| s.as_ref().map(Vec::len)) + .ok_or_else(|| io::Error::other("No valid shards found for parity encoding"))?; + + for shard in shards.iter_mut().skip(data_shards) { + if shard.is_none() { + *shard = Some(vec![0; shard_len]); + } + } + + let mut shard_refs: SmallVec<[&mut [u8]; 16]> = SmallVec::new(); + for (index, shard) in shards.iter_mut().enumerate() { + let shard = shard + .as_mut() + .ok_or_else(|| io::Error::other(format!("missing shard {index} after data reconstruction")))?; + if shard.len() != shard_len { + return Err(io::Error::other(format!( + "inconsistent shard length at index {index}: got {}, expected {}", + shard.len(), + shard_len + ))); + } + shard_refs.push(shard.as_mut_slice()); + } + + encode(shard_refs) } /// Erasure coding utility for data reliability using Reed-Solomon codes. @@ -392,7 +453,7 @@ impl Erasure { Ok(shards) } - /// Decode and reconstruct missing shards in-place. + /// Decode and reconstruct missing data shards in-place. /// /// # Arguments /// * `shards` - Mutable slice of optional shard data. Missing shards should be `None`. @@ -400,6 +461,25 @@ impl Erasure { /// # Returns /// Ok if reconstruction succeeds, error otherwise. pub fn decode_data(&self, shards: &mut [Option>]) -> io::Result<()> { + if self.parity_shards > 0 { + if self.uses_legacy { + if let Some(encoder) = self.legacy_encoder.as_ref() { + encoder.reconstruct_data(shards)?; + } else { + warn!("parity_shards > 0, uses_legacy but legacy_encoder is None"); + } + } else if let Some(encoder) = self.encoder.as_ref() { + encoder.reconstruct_data(shards)?; + } else { + warn!("parity_shards > 0, but encoder is None"); + } + } + + Ok(()) + } + + /// Decode and reconstruct missing data shards, then regenerate parity shards. + pub fn decode_data_and_parity(&self, shards: &mut [Option>]) -> io::Result<()> { if self.parity_shards > 0 { if self.uses_legacy { if let Some(encoder) = self.legacy_encoder.as_ref() { @@ -534,6 +614,131 @@ mod tests { use super::*; + fn optional_shards(shards: &[Bytes]) -> Vec>> { + shards.iter().map(|shard| Some(shard.to_vec())).collect() + } + + #[test] + fn decode_data_keeps_missing_parity_shard_unreconstructed() { + let erasure = Erasure::new(2, 2, 64); + let data = b"read decode should not rebuild parity"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut shards = optional_shards(&encoded); + let missing_parity = erasure.data_shards; + shards[missing_parity] = None; + + erasure.decode_data(&mut shards).expect("decode should succeed"); + + assert!(shards[missing_parity].is_none(), "read decode should leave parity missing"); + for index in 0..erasure.data_shards { + assert_eq!( + shards[index].as_deref(), + Some(encoded[index].as_ref()), + "data shard {index} should remain unchanged" + ); + } + } + + #[test] + fn legacy_decode_data_keeps_missing_parity_shard_unreconstructed() { + let erasure = Erasure::new_with_options(2, 2, 64, true); + let data = b"legacy read decode should not rebuild parity"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut shards = optional_shards(&encoded); + let missing_parity = erasure.data_shards + 1; + shards[missing_parity] = None; + + erasure.decode_data(&mut shards).expect("decode should succeed"); + + assert!(shards[missing_parity].is_none(), "legacy read decode should leave parity missing"); + for index in 0..erasure.data_shards { + assert_eq!( + shards[index].as_deref(), + Some(encoded[index].as_ref()), + "legacy data shard {index} should remain unchanged" + ); + } + } + + #[test] + fn decode_data_and_parity_leaves_complete_shards_unchanged() { + let erasure = Erasure::new(4, 2, 128); + let data = b"complete shards should not be changed"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let original = optional_shards(&encoded); + let mut shards = original.clone(); + + erasure + .decode_data_and_parity(&mut shards) + .expect("decode should succeed without missing shards"); + + assert_eq!(shards, original); + } + + #[test] + fn decode_data_and_parity_reconstructs_missing_parity_shard() { + let erasure = Erasure::new(2, 2, 64); + let data = b"parity shard must be rebuilt"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut shards = optional_shards(&encoded); + shards[2] = None; + + erasure + .decode_data_and_parity(&mut shards) + .expect("decode should rebuild parity"); + + for (index, shard) in shards.iter().enumerate() { + assert_eq!( + shard.as_deref(), + Some(encoded[index].as_ref()), + "shard {index} should match encoded source" + ); + } + } + + #[test] + fn decode_data_and_parity_reconstructs_missing_data_and_parity_shards() { + let erasure = Erasure::new(4, 2, 128); + let data = b"data and parity shards should both be reconstructed"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut shards = optional_shards(&encoded); + shards[1] = None; + shards[4] = None; + + erasure + .decode_data_and_parity(&mut shards) + .expect("decode should rebuild all missing shards"); + + for (index, shard) in shards.iter().enumerate() { + assert_eq!( + shard.as_deref(), + Some(encoded[index].as_ref()), + "shard {index} should match encoded source" + ); + } + } + + #[test] + fn legacy_decode_data_and_parity_reconstructs_missing_parity_shard() { + let erasure = Erasure::new_with_options(2, 2, 64, true); + let data = b"legacy parity shard must be rebuilt"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let mut shards = optional_shards(&encoded); + shards[3] = None; + + erasure + .decode_data_and_parity(&mut shards) + .expect("decode should rebuild parity"); + + for (index, shard) in shards.iter().enumerate() { + assert_eq!( + shard.as_deref(), + Some(encoded[index].as_ref()), + "shard {index} should match encoded source" + ); + } + } + #[test] fn test_shard_file_size_cases2() { let erasure = Erasure::new(12, 4, 1024 * 1024); diff --git a/crates/ecstore/src/erasure_coding/heal.rs b/crates/ecstore/src/erasure_coding/heal.rs index 422ae7dac..797d95bc4 100644 --- a/crates/ecstore/src/erasure_coding/heal.rs +++ b/crates/ecstore/src/erasure_coding/heal.rs @@ -49,6 +49,10 @@ impl super::Erasure { end_block += 1; } + let available_writers = writers.iter().filter(|w| w.is_some()).count(); + let write_quorum = available_writers.max(1); + let mut writers = MultiWriter::new(writers, write_quorum); + for _ in start_block..end_block { let (mut shards, errs) = reader.read().await; @@ -63,7 +67,7 @@ impl super::Erasure { } if self.parity_shards > 0 { - self.decode_data(&mut shards)?; + self.decode_data_and_parity(&mut shards)?; } let shards = shards @@ -71,15 +75,122 @@ impl super::Erasure { .map(|s| Bytes::from(s.unwrap_or_default())) .collect::>(); - // Calculate proper write quorum for heal operation - // For heal, we only write to disks that need healing, so write quorum should be - // the number of available writers (disks that need healing) - let available_writers = writers.iter().filter(|w| w.is_some()).count(); - let write_quorum = available_writers.max(1); // At least 1 writer must succeed - let mut writers = MultiWriter::new(writers, write_quorum); writers.write(shards).await?; } + writers.shutdown().await?; Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::erasure_coding::{CustomWriter, Erasure}; + use rustfs_utils::HashAlgorithm; + use std::io::Cursor; + + #[tokio::test] + async fn heal_reconstructs_missing_parity_shard() { + let erasure = Erasure::new(2, 2, 64); + let data = b"heal should write a rebuilt parity shard"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + let missing_parity = erasure.data_shards; + + let readers = encoded + .iter() + .enumerate() + .map(|(index, shard)| { + if index == missing_parity { + None + } else { + Some(BitrotReader::new( + Cursor::new(shard.to_vec()), + erasure.shard_size(), + HashAlgorithm::None, + false, + )) + } + }) + .collect::>(); + + let mut writers = (0..erasure.total_shard_count()) + .map(|index| { + if index == missing_parity { + Some(BitrotWriterWrapper::new( + CustomWriter::new_inline_buffer(), + erasure.shard_size(), + HashAlgorithm::None, + )) + } else { + None + } + }) + .collect::>(); + + erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect("heal should rebuild parity"); + + let healed = writers[missing_parity] + .take() + .expect("parity writer should remain") + .into_inline_data() + .expect("inline writer should retain data"); + assert_eq!(healed, encoded[missing_parity].to_vec()); + } + + #[tokio::test] + async fn heal_reconstructs_missing_data_shard_across_multiple_blocks() { + let erasure = Erasure::new(3, 2, 96); + let data = (0..erasure.block_size * 2 + 17) + .map(|index| (index % 251) as u8) + .collect::>(); + let encoded = erasure.encode_data(&data).expect("encode should succeed"); + let missing_data = 1; + + let readers = encoded + .iter() + .enumerate() + .map(|(index, shard)| { + if index == missing_data { + None + } else { + Some(BitrotReader::new( + Cursor::new(shard.to_vec()), + erasure.shard_size(), + HashAlgorithm::None, + false, + )) + } + }) + .collect::>(); + + let mut writers = (0..erasure.total_shard_count()) + .map(|index| { + if index == missing_data { + Some(BitrotWriterWrapper::new( + CustomWriter::new_inline_buffer(), + erasure.shard_size(), + HashAlgorithm::None, + )) + } else { + None + } + }) + .collect::>(); + + erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect("heal should rebuild data"); + + let healed = writers[missing_data] + .take() + .expect("data writer should remain") + .into_inline_data() + .expect("inline writer should retain data"); + assert_eq!(healed, encoded[missing_data].to_vec()); + } +}