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>

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();
+        // Heal is best-effort per target disk. A single replacement disk failing to
-        let write_quorum = available_writers.max(1);
+        // 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");
+    }
 }