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

@@ -106,7 +106,7 @@ impl LegacyReedSolomonEncoder {
         Ok(())
     }
 
-    fn reconstruct(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
+    fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> 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<Vec<u8>>]) -> io::Result<()> {
+        self.reconstruct_data(shards)?;
+        self.encode_parity(shards)
+    }
+
+    fn encode_parity(&self, shards: &mut [Option<Vec<u8>>]) -> 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<Vec<u8>>]) -> io::Result<()> {
+    /// Reconstruct missing data shards.
+    pub fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> 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<Vec<u8>>]) -> io::Result<()> {
+        self.reconstruct_data(shards)?;
+        self.encode_parity(shards)
+    }
+
+    fn encode_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
+        encode_parity_shards(shards, self.data_shards, self.parity_shards, |shards| self.encode(shards))
+    }
+}
+
+fn encode_parity_shards<F>(shards: &mut [Option<Vec<u8>>], 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<Vec<u8>>]) -> 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<Vec<u8>>]) -> 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<Option<Vec<u8>>> {
+        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);

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::<Vec<_>>();
 
-            // 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::<Vec<_>>();
+
+        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::<Vec<_>>();
+
+        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::<Vec<_>>();
+        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::<Vec<_>>();
+
+        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::<Vec<_>>();
+
+        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());
+    }
+}