ecstore update ec/disk/error

ecstore/src/erasure_coding/decode.rs

@@ -0,0 +1,263 @@
+use super::Erasure;
+use crate::disk::error::Error;
+use crate::disk::error_reduce::reduce_errs;
+use futures::future::join_all;
+use pin_project_lite::pin_project;
+use rustfs_rio::BitrotReader;
+use std::io;
+use std::io::ErrorKind;
+use tokio::io::AsyncWriteExt;
+use tracing::error;
+
+pin_project! {
+pub(crate) struct ParallelReader {
+    #[pin]
+    readers: Vec<Option<BitrotReader>>,
+    offset: usize,
+    shard_size: usize,
+    shard_file_size: usize,
+    data_shards: usize,
+    total_shards: usize,
+}
+}
+
+impl ParallelReader {
+    // readers传入前应处理disk错误，确保每个reader达到可用数量的BitrotReader
+    pub fn new(readers: Vec<Option<BitrotReader>>, e: Erasure, offset: usize, total_length: usize) -> Self {
+        let shard_size = e.shard_size();
+        let shard_file_size = e.shard_file_size(total_length);
+
+        let offset = (offset / e.block_size) * shard_size;
+
+        // 确保offset不超过shard_file_size
+
+        ParallelReader {
+            readers,
+            offset,
+            shard_size,
+            shard_file_size,
+            data_shards: e.data_shards,
+            total_shards: e.data_shards + e.parity_shards,
+        }
+    }
+}
+
+impl ParallelReader {
+    pub async fn read(&mut self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
+        // if self.readers.len() != self.total_shards {
+        //     return Err(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers"));
+        // }
+
+        let shard_size = if self.offset + self.shard_size > self.shard_file_size {
+            self.shard_file_size - self.offset
+        } else {
+            self.shard_size
+        };
+
+        if shard_size == 0 {
+            return (vec![None; self.readers.len()], vec![None; self.readers.len()]);
+        }
+
+        // 使用并发读取所有分片
+
+        let read_futs: Vec<_> = self
+            .readers
+            .iter_mut()
+            .enumerate()
+            .map(|(i, opt_reader)| {
+                if let Some(reader) = opt_reader.as_mut() {
+                    let mut buf = vec![0u8; shard_size];
+                    // 需要move i, buf
+                    Some(async move {
+                        match reader.read(&mut buf).await {
+                            Ok(n) => {
+                                buf.truncate(n);
+                                (i, Ok(buf))
+                            }
+                            Err(e) => (i, Err(Error::from(e))),
+                        }
+                    })
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        // 过滤掉None，join_all
+        let mut results = join_all(read_futs.into_iter().flatten()).await;
+
+        let mut shards: Vec<Option<Vec<u8>>> = vec![None; self.readers.len()];
+        let mut errs = vec![None; self.readers.len()];
+        for (i, shard) in results.drain(..) {
+            match shard {
+                Ok(data) => {
+                    if !data.is_empty() {
+                        shards[i] = Some(data);
+                    }
+                }
+                Err(e) => {
+                    error!("Error reading shard {}: {}", i, e);
+                    errs[i] = Some(e);
+                }
+            }
+        }
+
+        self.offset += shard_size;
+
+        (shards, errs)
+    }
+
+    pub fn can_decode(&self, shards: &[Option<Vec<u8>>]) -> bool {
+        shards.iter().filter(|s| s.is_some()).count() >= self.data_shards
+    }
+}
+
+/// 获取数据块总长度
+fn get_data_block_len(shards: &[Option<Vec<u8>>], data_blocks: usize) -> usize {
+    let mut size = 0;
+    for shard in shards.iter().take(data_blocks).flatten() {
+        size += shard.len();
+    }
+
+    size
+}
+
+/// 将编码块中的数据块写入目标，支持 offset 和 length
+async fn write_data_blocks<W>(
+    writer: &mut W,
+    en_blocks: &[Option<Vec<u8>>],
+    data_blocks: usize,
+    mut offset: usize,
+    length: usize,
+) -> std::io::Result<usize>
+where
+    W: tokio::io::AsyncWrite + Send + Sync + Unpin,
+{
+    if get_data_block_len(en_blocks, data_blocks) < length {
+        return Err(io::Error::new(ErrorKind::UnexpectedEof, "Not enough data blocks to write"));
+    }
+
+    let mut total_written = 0;
+    let mut write_left = length;
+
+    for block_op in &en_blocks[..data_blocks] {
+        if block_op.is_none() {
+            return Err(io::Error::new(ErrorKind::UnexpectedEof, "Missing data block"));
+        }
+
+        let block = block_op.as_ref().unwrap();
+
+        if offset >= block.len() {
+            offset -= block.len();
+            continue;
+        }
+
+        let block_slice = &block[offset..];
+        offset = 0;
+
+        if write_left < block.len() {
+            writer.write_all(&block_slice[..write_left]).await?;
+
+            total_written += write_left;
+            break;
+        }
+
+        let n = block_slice.len();
+
+        writer.write_all(block_slice).await?;
+
+        write_left -= n;
+
+        total_written += n;
+    }
+
+    Ok(total_written)
+}
+
+impl Erasure {
+    pub async fn decode<W>(
+        &self,
+        writer: &mut W,
+        readers: Vec<Option<BitrotReader>>,
+        offset: usize,
+        length: usize,
+        total_length: usize,
+    ) -> (usize, Option<std::io::Error>)
+    where
+        W: tokio::io::AsyncWrite + Send + Sync + Unpin + 'static,
+    {
+        if readers.len() != self.data_shards + self.parity_shards {
+            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")));
+        }
+
+        if offset + length > total_length {
+            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")));
+        }
+
+        let mut ret_err = None;
+
+        if length == 0 {
+            return (0, ret_err);
+        }
+
+        let mut written = 0;
+
+        let mut reader = ParallelReader::new(readers, self.clone(), offset, total_length);
+
+        let start = offset / self.block_size;
+        let end = (offset + length) / self.block_size;
+
+        for i in start..=end {
+            let (block_offset, block_length) = if start == end {
+                (offset % self.block_size, length)
+            } else if i == start {
+                (offset % self.block_size, self.block_size - (offset % self.block_size))
+            } else if i == end {
+                (0, (offset + length) % self.block_size)
+            } else {
+                (0, self.block_size)
+            };
+
+            if block_length == 0 {
+                break;
+            }
+
+            let (mut shards, errs) = reader.read().await;
+
+            if ret_err.is_none() {
+                if let (_, Some(err)) = reduce_errs(&errs, &[]) {
+                    if err == Error::FileNotFound || err == Error::FileCorrupt {
+                        ret_err = Some(err.into());
+                    }
+                }
+            }
+
+            if !reader.can_decode(&shards) {
+                ret_err = Some(Error::ErasureReadQuorum.into());
+                break;
+            }
+
+            // Decode the shards
+            if let Err(e) = self.decode_data(&mut shards) {
+                ret_err = Some(e);
+                break;
+            }
+
+            let n = match write_data_blocks(writer, &shards, self.data_shards, block_offset, block_length).await {
+                Ok(n) => n,
+                Err(e) => {
+                    ret_err = Some(e);
+                    break;
+                }
+            };
+
+            written += n;
+        }
+
+        if written < length {
+            ret_err = Some(Error::LessData.into());
+        }
+
+        (written, ret_err)
+    }
+}

ecstore/src/erasure_coding/encode.rs

@@ -0,0 +1,139 @@
+use bytes::Bytes;
+use rustfs_rio::BitrotWriter;
+use rustfs_rio::Reader;
+// use std::io::Cursor;
+// use std::mem;
+use super::Erasure;
+use crate::disk::error::Error;
+use crate::disk::error_reduce::count_errs;
+use crate::disk::error_reduce::{reduce_write_quorum_errs, OBJECT_OP_IGNORED_ERRS};
+use std::sync::Arc;
+use std::vec;
+use tokio::sync::mpsc;
+
+pub(crate) struct MultiWriter<'a> {
+    writers: &'a mut [Option<BitrotWriter>],
+    write_quorum: usize,
+    errs: Vec<Option<Error>>,
+}
+
+impl<'a> MultiWriter<'a> {
+    pub fn new(writers: &'a mut [Option<BitrotWriter>], write_quorum: usize) -> Self {
+        let length = writers.len();
+        MultiWriter {
+            writers,
+            write_quorum,
+            errs: vec![None; length],
+        }
+    }
+
+    #[allow(clippy::needless_range_loop)]
+    pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
+        for i in 0..self.writers.len() {
+            if self.errs[i].is_some() {
+                continue; // Skip if we already have an error for this writer
+            }
+
+            let writer_opt = &mut self.writers[i];
+            let shard = &data[i];
+
+            if let Some(writer) = writer_opt {
+                match writer.write(shard).await {
+                    Ok(n) => {
+                        if n < shard.len() {
+                            self.errs[i] = Some(Error::ShortWrite);
+                            self.writers[i] = None; // Mark as failed
+                        } else {
+                            self.errs[i] = None;
+                        }
+                    }
+                    Err(e) => {
+                        self.errs[i] = Some(Error::from(e));
+                    }
+                }
+            } else {
+                self.errs[i] = Some(Error::DiskNotFound);
+            }
+        }
+
+        let nil_count = self.errs.iter().filter(|&e| e.is_none()).count();
+        if nil_count > self.write_quorum {
+            return Ok(());
+        }
+
+        if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!(
+                    "Failed to write data: {} (offline-disks={}/{})",
+                    write_err,
+                    count_errs(&self.errs, &Error::DiskNotFound),
+                    self.writers.len()
+                ),
+            ));
+        }
+
+        Err(std::io::Error::other(format!(
+            "Failed to write data:  (offline-disks={}/{})",
+            count_errs(&self.errs, &Error::DiskNotFound),
+            self.writers.len()
+        )))
+    }
+}
+
+impl Erasure {
+    pub async fn encode<R>(
+        self: Arc<Self>,
+        mut reader: R,
+        writers: &mut [Option<BitrotWriter>],
+        quorum: usize,
+    ) -> std::io::Result<(R, usize)>
+    where
+        R: Reader + Send + Sync + Unpin + 'static,
+    {
+        let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(8);
+
+        let task = tokio::spawn(async move {
+            let block_size = self.block_size;
+            let mut total = 0;
+            loop {
+                let mut buf = vec![0u8; block_size];
+                match rustfs_utils::read_full(&mut reader, &mut buf).await {
+                    Ok(n) if n > 0 => {
+                        total += n;
+                        let res = self.encode_data(&buf[..n])?;
+                        if let Err(err) = tx.send(res).await {
+                            return Err(std::io::Error::new(
+                                std::io::ErrorKind::Other,
+                                format!("Failed to send encoded data : {}", err),
+                            ));
+                        }
+                    }
+                    Ok(_) => break,
+                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                        break;
+                    }
+                    Err(e) => {
+                        return Err(e);
+                    }
+                }
+                buf.clear();
+            }
+
+            Ok((reader, total))
+        });
+
+        let mut writers = MultiWriter::new(writers, quorum);
+
+        while let Some(block) = rx.recv().await {
+            if block.is_empty() {
+                break;
+            }
+            writers.write(block).await?;
+        }
+
+        let (reader, total) = task.await??;
+
+        Ok((reader, total))
+    }
+}

ecstore/src/erasure_coding/erasure.rs

@@ -0,0 +1,433 @@
+use bytes::{Bytes, BytesMut};
+use reed_solomon_erasure::galois_8::ReedSolomon;
+// use rustfs_rio::Reader;
+use smallvec::SmallVec;
+use std::io;
+use std::io::ErrorKind;
+use tracing::error;
+use tracing::warn;
+use uuid::Uuid;
+
+/// Erasure coding utility for data reliability using Reed-Solomon codes.
+///
+/// This struct provides encoding and decoding of data into data and parity shards.
+/// It supports splitting data into multiple shards, generating parity for fault tolerance,
+/// and reconstructing lost shards.
+///
+/// # Fields
+/// - `data_shards`: Number of data shards.
+/// - `parity_shards`: Number of parity shards.
+/// - `encoder`: Optional ReedSolomon encoder instance.
+/// - `block_size`: Block size for each shard.
+/// - `_id`: Unique identifier for the erasure instance.
+/// - `_buf`: Internal buffer for block operations.
+///
+/// # Example
+/// ```
+/// use erasure_coding::Erasure;
+/// let erasure = Erasure::new(4, 2, 8);
+/// let data = b"hello world";
+/// let shards = erasure.encode_data(data).unwrap();
+/// // Simulate loss and recovery...
+/// ```
+
+#[derive(Default, Clone)]
+pub struct Erasure {
+    pub data_shards: usize,
+    pub parity_shards: usize,
+    encoder: Option<ReedSolomon>,
+    pub block_size: usize,
+    _id: Uuid,
+    _buf: Vec<u8>,
+}
+
+impl Erasure {
+    /// Create a new Erasure instance.
+    ///
+    /// # Arguments
+    /// * `data_shards` - Number of data shards.
+    /// * `parity_shards` - Number of parity shards.
+    /// * `block_size` - Block size for each shard.
+    pub fn new(data_shards: usize, parity_shards: usize, block_size: usize) -> Self {
+        let encoder = if parity_shards > 0 {
+            Some(ReedSolomon::new(data_shards, parity_shards).unwrap())
+        } else {
+            None
+        };
+
+        Erasure {
+            data_shards,
+            parity_shards,
+            block_size,
+            encoder,
+            _id: Uuid::new_v4(),
+            _buf: vec![0u8; block_size],
+        }
+    }
+
+    /// Encode data into data and parity shards.
+    ///
+    /// # Arguments
+    /// * `data` - The input data to encode.
+    ///
+    /// # Returns
+    /// A vector of encoded shards as `Bytes`.
+    #[tracing::instrument(level = "info", skip_all, fields(data_len=data.len()))]
+    pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
+        // let shard_size = self.shard_size();
+        // let total_size = shard_size * self.total_shard_count();
+
+        // 数据切片数量
+        let per_shard_size = data.len().div_ceil(self.data_shards);
+        // 总需求大小
+        let need_total_size = per_shard_size * self.total_shard_count();
+
+        // Create a new buffer with the required total length for all shards
+        let mut data_buffer = BytesMut::with_capacity(need_total_size);
+
+        // Copy source data
+        data_buffer.extend_from_slice(data);
+        data_buffer.resize(need_total_size, 0u8);
+
+        {
+            // EC encode, the result will be written into data_buffer
+            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();
+
+            // Only do EC if parity_shards > 0
+            if self.parity_shards > 0 {
+                if let Some(encoder) = self.encoder.as_ref() {
+                    encoder.encode(data_slices).map_err(|e| {
+                        error!("encode data error: {:?}", e);
+                        io::Error::new(ErrorKind::Other, format!("encode data error {:?}", e))
+                    })?;
+                } else {
+                    warn!("parity_shards > 0, but encoder is None");
+                }
+            }
+        }
+
+        // Zero-copy split, all shards reference data_buffer
+        let mut data_buffer = data_buffer.freeze();
+        let mut shards = Vec::with_capacity(self.total_shard_count());
+        for _ in 0..self.total_shard_count() {
+            let shard = data_buffer.split_to(per_shard_size);
+            shards.push(shard);
+        }
+
+        Ok(shards)
+    }
+
+    /// Decode and reconstruct missing shards in-place.
+    ///
+    /// # Arguments
+    /// * `shards` - Mutable slice of optional shard data. Missing shards should be `None`.
+    ///
+    /// # 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 let Some(encoder) = self.encoder.as_ref() {
+                encoder.reconstruct(shards).map_err(|e| {
+                    error!("decode data error: {:?}", e);
+                    io::Error::new(ErrorKind::Other, format!("decode data error {:?}", e))
+                })?;
+            } else {
+                warn!("parity_shards > 0, but encoder is None");
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Get the total number of shards (data + parity).
+    pub fn total_shard_count(&self) -> usize {
+        self.data_shards + self.parity_shards
+    }
+    // /// Calculate the shard size and total size for a given data size.
+    // // Returns (shard_size, total_size) for the given data size
+    // fn need_size(&self, data_size: usize) -> (usize, usize) {
+    //     let shard_size = self.shard_size(data_size);
+    //     (shard_size, shard_size * (self.total_shard_count()))
+    // }
+
+    /// Calculate the size of each shard.
+    pub fn shard_size(&self) -> usize {
+        self.block_size.div_ceil(self.data_shards)
+    }
+    /// Calculate the total erasure file size for a given original size.
+    // Returns the final erasure size from the original size
+    pub fn shard_file_size(&self, total_length: usize) -> usize {
+        if total_length == 0 {
+            return 0;
+        }
+
+        let num_shards = total_length / self.block_size;
+        let last_block_size = total_length % self.block_size;
+        let last_shard_size = last_block_size.div_ceil(self.data_shards);
+        num_shards * self.shard_size() + last_shard_size
+    }
+
+    /// Calculate the offset in the erasure file where reading begins.
+    // Returns the offset in the erasure file where reading begins
+    pub fn shard_file_offset(&self, start_offset: usize, length: usize, total_length: usize) -> usize {
+        let shard_size = self.shard_size();
+        let shard_file_size = self.shard_file_size(total_length);
+        let end_shard = (start_offset + length) / self.block_size;
+        let mut till_offset = end_shard * shard_size + shard_size;
+        if till_offset > shard_file_size {
+            till_offset = shard_file_size;
+        }
+
+        till_offset
+    }
+
+    /// Encode all data from a rustfs_rio::Reader in blocks, calling an async callback for each encoded block.
+    /// This method is async and returns the reader and total bytes read after all blocks are processed.
+    ///
+    /// # Arguments
+    /// * `reader` - A rustfs_rio::Reader to read data from.
+    /// * `mut on_block` - Async callback: FnMut(Result<Vec<Bytes>, std::io::Error>) -> Future<Output=Result<(), E>> + Send
+    ///
+    /// # Returns
+    /// Result<(reader, total_bytes_read), E> after all data has been processed or on callback error.
+    pub async fn encode_stream_callback_async<F, Fut, E, R>(
+        self: std::sync::Arc<Self>,
+        reader: &mut R,
+        mut on_block: F,
+    ) -> Result<usize, E>
+    where
+        R: rustfs_rio::Reader + Send + Sync + Unpin,
+        F: FnMut(std::io::Result<Vec<Bytes>>) -> Fut + Send,
+        Fut: std::future::Future<Output = Result<(), E>> + Send,
+    {
+        let block_size = self.block_size;
+        let mut total = 0;
+        loop {
+            let mut buf = vec![0u8; block_size];
+            match rustfs_utils::read_full(&mut *reader, &mut buf).await {
+                Ok(n) if n > 0 => {
+                    total += n;
+                    let res = self.encode_data(&buf[..n]);
+                    on_block(res).await?
+                }
+                Ok(_) => break,
+                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                    break;
+                }
+                Err(e) => {
+                    on_block(Err(e)).await?;
+                    break;
+                }
+            }
+            buf.clear();
+        }
+        Ok(total)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_shard_file_size_cases() {
+        let erasure = Erasure::new(4, 2, 8);
+
+        // Case 1: total_length == 0
+        assert_eq!(erasure.shard_file_size(0), 0);
+
+        // Case 2: total_length < block_size
+        assert_eq!(erasure.shard_file_size(5), 2); // 5 div_ceil 4 = 2
+
+        // Case 3: total_length == block_size
+        assert_eq!(erasure.shard_file_size(8), 2);
+
+        // Case 4: total_length > block_size, not aligned
+        assert_eq!(erasure.shard_file_size(13), 4); // 8/8=1, last=5, 5 div_ceil 4=2, 1*2+2=4
+
+        // Case 5: total_length > block_size, aligned
+        assert_eq!(erasure.shard_file_size(16), 4); // 16/8=2, last=0, 2*2+0=4
+
+        assert_eq!(erasure.shard_file_size(1248739), 312185); // 1248739/8=156092, last=3, 3 div_ceil 4=1, 156092*2+1=312185
+
+        assert_eq!(erasure.shard_file_size(43), 11); // 43/8=5, last=3, 3 div_ceil 4=1, 5*2+1=11
+    }
+
+    #[test]
+    fn test_encode_decode_roundtrip() {
+        let data_shards = 4;
+        let parity_shards = 2;
+        let block_size = 8;
+        let erasure = Erasure::new(data_shards, parity_shards, block_size);
+        // let data = b"hello erasure coding!";
+        let data = b"channel async callback test data!";
+        let shards = erasure.encode_data(data).unwrap();
+        // Simulate the loss of one shard
+        let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
+        shards_opt[2] = None;
+        // Decode
+        erasure.decode_data(&mut shards_opt).unwrap();
+        // Recover original data
+        let mut recovered = Vec::new();
+        for shard in shards_opt.iter().take(data_shards) {
+            recovered.extend_from_slice(shard.as_ref().unwrap());
+        }
+        recovered.truncate(data.len());
+        assert_eq!(&recovered, data);
+    }
+
+    #[test]
+    fn test_encode_all_zero_data() {
+        let data_shards = 3;
+        let parity_shards = 2;
+        let block_size = 6;
+        let erasure = Erasure::new(data_shards, parity_shards, block_size);
+        let data = vec![0u8; block_size];
+        let shards = erasure.encode_data(&data).unwrap();
+        assert_eq!(shards.len(), data_shards + parity_shards);
+        let total_len: usize = shards.iter().map(|b| b.len()).sum();
+        assert_eq!(total_len, erasure.shard_size() * (data_shards + parity_shards));
+    }
+
+    #[test]
+    fn test_shard_size_and_file_size() {
+        let erasure = Erasure::new(4, 2, 8);
+        assert_eq!(erasure.shard_file_size(33), 9);
+        assert_eq!(erasure.shard_file_size(0), 0);
+    }
+
+    #[test]
+    fn test_shard_file_offset() {
+        let erasure = Erasure::new(4, 2, 8);
+        let offset = erasure.shard_file_offset(0, 16, 32);
+        assert!(offset > 0);
+    }
+
+    #[test]
+    fn test_encode_decode_large_1m() {
+        // Test encoding and decoding 1MB data, simulating the loss of 2 shards
+        let data_shards = 6;
+        let parity_shards = 3;
+        let block_size = 128 * 1024; // 128KB
+        let erasure = Erasure::new(data_shards, parity_shards, block_size);
+        let data = vec![0x5Au8; 1024 * 1024]; // 1MB fixed content
+        let shards = erasure.encode_data(&data).unwrap();
+        // Simulate the loss of 2 shards
+        let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
+        shards_opt[1] = None;
+        shards_opt[7] = None;
+        // Decode
+        erasure.decode_data(&mut shards_opt).unwrap();
+        // Recover original data
+        let mut recovered = Vec::new();
+        for shard in shards_opt.iter().take(data_shards) {
+            recovered.extend_from_slice(shard.as_ref().unwrap());
+        }
+        recovered.truncate(data.len());
+        assert_eq!(&recovered, &data);
+    }
+
+    #[tokio::test]
+    async fn test_encode_stream_callback_async_error_propagation() {
+        use std::sync::Arc;
+        use tokio::io::BufReader;
+        use tokio::sync::mpsc;
+        let data_shards = 3;
+        let parity_shards = 3;
+        let block_size = 8;
+        let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));
+        let data = b"async stream callback error propagation!123";
+        let mut rio_reader = BufReader::new(&data[..]);
+        let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(8);
+        let erasure_clone = erasure.clone();
+        let mut call_count = 0;
+        let handle = tokio::spawn(async move {
+            let result = erasure_clone
+                .encode_stream_callback_async::<_, _, &'static str, _>(&mut rio_reader, move |res| {
+                    let tx = tx.clone();
+                    call_count += 1;
+                    async move {
+                        if call_count == 2 {
+                            Err("user error")
+                        } else {
+                            let shards = res.unwrap();
+                            tx.send(shards).await.unwrap();
+                            Ok(())
+                        }
+                    }
+                })
+                .await;
+            assert!(result.is_err());
+            assert_eq!(result.unwrap_err(), "user error");
+        });
+        let mut all_blocks = Vec::new();
+        while let Some(block) = rx.recv().await {
+            println!("Received block: {:?}", block[0].len());
+            all_blocks.push(block);
+        }
+        handle.await.unwrap();
+        // 只处理了第一个 block
+        assert_eq!(all_blocks.len(), 1);
+        // 对第一个 block 使用 decode_data 修复并校验
+        let block = &all_blocks[0];
+        let mut shards_opt: Vec<Option<Vec<u8>>> = block.iter().map(|b| Some(b.to_vec())).collect();
+        // 模拟丢失一个分片
+        shards_opt[0] = None;
+        erasure.decode_data(&mut shards_opt).unwrap();
+
+        let mut recovered = Vec::new();
+        for shard in shards_opt.iter().take(data_shards) {
+            recovered.extend_from_slice(shard.as_ref().unwrap());
+        }
+        // 只恢复第一个 block 的原始数据
+        let block_data_len = std::cmp::min(block_size, data.len());
+        recovered.truncate(block_data_len);
+        assert_eq!(&recovered, &data[..block_data_len]);
+    }
+
+    #[tokio::test]
+    async fn test_encode_stream_callback_async_channel_decode() {
+        use std::sync::Arc;
+        use tokio::io::BufReader;
+        use tokio::sync::mpsc;
+        let data_shards = 4;
+        let parity_shards = 2;
+        let block_size = 8;
+        let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));
+        let data = b"channel async callback test data!";
+        let mut rio_reader = BufReader::new(&data[..]);
+        let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(8);
+        let erasure_clone = erasure.clone();
+        let handle = tokio::spawn(async move {
+            erasure_clone
+                .encode_stream_callback_async::<_, _, (), _>(&mut rio_reader, move |res| {
+                    let tx = tx.clone();
+                    async move {
+                        let shards = res.unwrap();
+                        tx.send(shards).await.unwrap();
+                        Ok(())
+                    }
+                })
+                .await
+                .unwrap();
+        });
+        let mut all_blocks = Vec::new();
+        while let Some(block) = rx.recv().await {
+            all_blocks.push(block);
+        }
+        handle.await.unwrap();
+        // 对每个 block，模拟丢失一个分片并恢复
+        let mut recovered = Vec::new();
+        for block in &all_blocks {
+            let mut shards_opt: Vec<Option<Vec<u8>>> = block.iter().map(|b| Some(b.to_vec())).collect();
+            // 模拟丢失一个分片
+            shards_opt[0] = None;
+            erasure.decode_data(&mut shards_opt).unwrap();
+            for shard in shards_opt.iter().take(data_shards) {
+                recovered.extend_from_slice(shard.as_ref().unwrap());
+            }
+        }
+        recovered.truncate(data.len());
+        assert_eq!(&recovered, data);
+    }
+}

ecstore/src/erasure_coding/heal.rs

@@ -0,0 +1,56 @@
+use super::decode::ParallelReader;
+use crate::disk::error::{Error, Result};
+use crate::erasure_coding::encode::MultiWriter;
+use bytes::Bytes;
+use rustfs_rio::BitrotReader;
+use rustfs_rio::BitrotWriter;
+use tracing::info;
+
+impl super::Erasure {
+    pub async fn heal(
+        &self,
+        writers: &mut [Option<BitrotWriter>],
+        readers: Vec<Option<BitrotReader>>,
+        total_length: usize,
+        _prefer: &[bool],
+    ) -> Result<()> {
+        info!(
+            "Erasure heal, writers len: {}, readers len: {}, total_length: {}",
+            writers.len(),
+            readers.len(),
+            total_length
+        );
+        if writers.len() != self.parity_shards + self.data_shards {
+            return Err(Error::other("invalid argument"));
+        }
+        let mut reader = ParallelReader::new(readers, self.clone(), 0, total_length);
+
+        let start_block = 0;
+        let mut end_block = total_length / self.block_size;
+        if total_length % self.block_size != 0 {
+            end_block += 1;
+        }
+
+        for _ in start_block..end_block {
+            let (mut shards, errs) = reader.read().await;
+
+            if errs.iter().filter(|e| e.is_none()).count() < self.data_shards {
+                return Err(Error::other(format!("can not reconstruct data: not enough data shards {:?}", errs)));
+            }
+
+            if self.parity_shards > 0 {
+                self.decode_data(&mut shards)?;
+            }
+
+            let shards = shards
+                .into_iter()
+                .map(|s| Bytes::from(s.unwrap_or_default()))
+                .collect::<Vec<_>>();
+
+            let mut writers = MultiWriter::new(writers, self.data_shards);
+            writers.write(shards).await?;
+        }
+
+        Ok(())
+    }
+}

ecstore/src/erasure_coding/mod.rs

@@ -0,0 +1,6 @@
+pub mod decode;
+pub mod encode;
+pub mod erasure;
+pub mod heal;
+
+pub use erasure::Erasure;