rustfs/crates/ecstore/src/erasure/coding/erasure.rs

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Erasure coding implementation using reed-solomon-erasure (GF(2^8)).
//! Supports legacy (reed-solomon-simd) for reading/healing old-version files.
//!

use bytes::{Bytes, BytesMut};
use reed_solomon_erasure::galois_8::ReedSolomon;
use reed_solomon_simd;
use smallvec::SmallVec;
use std::io;
use tokio::io::AsyncRead;
use tracing::warn;
use uuid::Uuid;

const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;

/// Errors returned when constructing an [`Erasure`] codec.
#[derive(Debug, thiserror::Error)]
pub enum ErasureConstructionError {
    /// Erasure coding requires at least one data shard.
    #[error("data_shards must be greater than zero")]
    ZeroDataShards,

    /// Erasure coding requires a non-zero logical block size.
    #[error("block_size must be greater than zero")]
    ZeroBlockSize,

    /// The total shard count cannot be represented by `usize`.
    #[error("data_shards ({data_shards}) + parity_shards ({parity_shards}) overflows usize")]
    ShardCountOverflow { data_shards: usize, parity_shards: usize },

    /// The modern GF(2^8) codec cannot represent this shard configuration.
    #[error("modern codec does not support {data_shards} data shards and {parity_shards} parity shards")]
    UnsupportedModernShardCount { data_shards: usize, parity_shards: usize },

    /// The legacy SIMD codec cannot represent this shard configuration.
    #[error("legacy codec does not support {data_shards} data shards and {parity_shards} parity shards")]
    UnsupportedLegacyShardCount { data_shards: usize, parity_shards: usize },

    /// The modern encoder rejected dimensions that passed the preflight checks.
    #[error("failed to construct modern Reed-Solomon encoder")]
    ModernEncoder {
        #[source]
        source: reed_solomon_erasure::Error,
    },

    /// The legacy encoder wrapper failed to initialize.
    #[error("failed to construct legacy Reed-Solomon encoder")]
    LegacyEncoder {
        #[source]
        source: io::Error,
    },
}

#[derive(Debug, thiserror::Error)]
#[error("{source}")]
struct ErasureConstructionIoSource {
    #[source]
    source: ErasureConstructionError,
}

impl ErasureConstructionError {
    pub(crate) fn into_io_error(self) -> io::Error {
        // `io::Error::source` forwards to the wrapped error's source rather
        // than exposing the wrapped error itself. This adapter keeps the
        // construction error as an observable link in the source chain.
        io::Error::other(ErasureConstructionIoSource { source: self })
    }
}

/// Legacy calc_shard_size formula: (block_size.div_ceil(data_shards) + 1) & !1
/// Matches main branch and filemeta::ErasureInfo for old-version files.
pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize {
    (block_size.div_ceil(data_shards) + 1) & !1
}

/// Reed-Solomon encoder for legacy (main branch) format using reed-solomon-simd.
/// Used when decoding/encoding files with uses_legacy_checksum == true.
struct LegacyReedSolomonEncoder {
    data_shards: usize,
    parity_shards: usize,
    encoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
    decoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
}

impl Clone for LegacyReedSolomonEncoder {
    fn clone(&self) -> Self {
        Self {
            data_shards: self.data_shards,
            parity_shards: self.parity_shards,
            encoder_cache: std::sync::RwLock::new(None),
            decoder_cache: std::sync::RwLock::new(None),
        }
    }
}

impl LegacyReedSolomonEncoder {
    fn new(_data_shards: usize, _parity_shards: usize) -> io::Result<Self> {
        Ok(Self {
            data_shards: _data_shards,
            parity_shards: _parity_shards,
            encoder_cache: std::sync::RwLock::new(None),
            decoder_cache: std::sync::RwLock::new(None),
        })
    }

    fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
        let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
        if shards_vec.is_empty() {
            return Ok(());
        }
        let shard_len = shards_vec[0].len();
        let mut encoder = {
            let mut cache_guard = self
                .encoder_cache
                .write()
                .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?;
            match cache_guard.take() {
                Some(mut cached) => {
                    if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() {
                        reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
                            .map_err(|e| io::Error::other(format!("Failed to create SIMD encoder: {e:?}")))?
                    } else {
                        cached
                    }
                }
                None => reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
                    .map_err(|e| io::Error::other(format!("Failed to create SIMD encoder: {e:?}")))?,
            }
        };
        for (i, shard) in shards_vec.iter().enumerate().take(self.data_shards) {
            encoder
                .add_original_shard(shard)
                .map_err(|e| io::Error::other(format!("Failed to add shard {i}: {e:?}")))?;
        }
        let result = encoder
            .encode()
            .map_err(|e| io::Error::other(format!("SIMD encoding failed: {e:?}")))?;
        for (i, recovery_shard) in result.recovery_iter().enumerate() {
            if i + self.data_shards < shards_vec.len() {
                shards_vec[i + self.data_shards].copy_from_slice(recovery_shard);
            }
        }
        drop(result);
        *self
            .encoder_cache
            .write()
            .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder);
        Ok(())
    }

    fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
        if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? {
            return Ok(());
        }

        let shard_len = shards
            .iter()
            .find_map(|s| s.as_ref().map(|v| v.len()))
            .ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?;

        let mut decoder = {
            let mut cache_guard = self
                .decoder_cache
                .write()
                .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?;

            match cache_guard.take() {
                Some(mut cached_decoder) => {
                    if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) {
                        warn!("Failed to reset SIMD decoder: {:?}, creating new one", e);
                        reed_solomon_simd::ReedSolomonDecoder::new(self.data_shards, self.parity_shards, shard_len)
                            .map_err(|e| io::Error::other(format!("Failed to create SIMD decoder: {e:?}")))?
                    } else {
                        cached_decoder
                    }
                }
                None => reed_solomon_simd::ReedSolomonDecoder::new(self.data_shards, self.parity_shards, shard_len)
                    .map_err(|e| io::Error::other(format!("Failed to create SIMD decoder: {e:?}")))?,
            }
        };

        for (i, shard_opt) in shards.iter().enumerate() {
            if let Some(shard) = shard_opt {
                if i < self.data_shards {
                    decoder
                        .add_original_shard(i, shard)
                        .map_err(|e| io::Error::other(format!("Failed to add original shard for reconstruction: {e:?}")))?;
                } else {
                    let recovery_idx = i - self.data_shards;
                    decoder
                        .add_recovery_shard(recovery_idx, shard)
                        .map_err(|e| io::Error::other(format!("Failed to add recovery shard for reconstruction: {e:?}")))?;
                }
            }
        }

        let result = decoder
            .decode()
            .map_err(|e| io::Error::other(format!("SIMD decode error: {e:?}")))?;

        for (i, shard_opt) in shards.iter_mut().enumerate() {
            if shard_opt.is_none() && i < self.data_shards {
                for (restored_index, restored_data) in result.restored_original_iter() {
                    if restored_index == i {
                        *shard_opt = Some(restored_data.to_vec());
                        break;
                    }
                }
            }
        }

        drop(result);

        *self
            .decoder_cache
            .write()
            .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder);

        Ok(())
    }

    fn reconstruct(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
        self.reconstruct_data(shards)?;
        self.encode_parity(shards)
    }

    fn verify(&self, shards: &[&[u8]]) -> io::Result<bool> {
        let expected_shards = self.data_shards + self.parity_shards;
        if shards.len() != expected_shards {
            return Err(io::Error::other(format!(
                "invalid shard count: got {}, expected {}",
                shards.len(),
                expected_shards
            )));
        }
        if shards.iter().all(|shard| shard.is_empty()) {
            return Ok(true);
        }

        let mut expected = shards.iter().map(|shard| Some(shard.to_vec())).collect::<Vec<_>>();
        self.encode_parity(&mut expected)?;

        for index in self.data_shards..expected_shards {
            let Some(expected_parity) = expected[index].as_ref() else {
                return Err(io::Error::other(format!("missing parity shard {index} after verification encode")));
            };
            if expected_parity.as_slice() != shards[index] {
                return Ok(false);
            }
        }

        Ok(true)
    }

    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
pub struct ReedSolomonEncoder {
    data_shards: usize,
    parity_shards: usize,
    encoder: Option<ReedSolomon>,
}

impl Clone for ReedSolomonEncoder {
    fn clone(&self) -> Self {
        Self {
            data_shards: self.data_shards,
            parity_shards: self.parity_shards,
            encoder: self.encoder.clone(),
        }
    }
}

impl ReedSolomonEncoder {
    fn try_new_typed(data_shards: usize, parity_shards: usize) -> Result<Self, reed_solomon_erasure::Error> {
        let encoder = if parity_shards > 0 {
            Some(ReedSolomon::new(data_shards, parity_shards)?)
        } else {
            None
        };

        Ok(ReedSolomonEncoder {
            data_shards,
            parity_shards,
            encoder,
        })
    }

    /// Create a new Reed-Solomon encoder with specified data and parity shards.
    pub fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
        Self::try_new_typed(data_shards, parity_shards)
            .map_err(|error| io::Error::other(format!("Failed to create Reed-Solomon encoder: {error:?}")))
    }

    /// Encode data shards with parity.
    pub fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
        let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
        if shards_vec.is_empty() {
            return Ok(());
        }

        if let Some(ref rs) = self.encoder {
            rs.encode(&mut shards_vec)
                .map_err(|e| io::Error::other(format!("Reed-Solomon encode failed: {e:?}")))
        } else {
            Ok(())
        }
    }

    /// Reconstruct missing data shards.
    pub fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
        if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? {
            return Ok(());
        }

        if let Some(ref rs) = self.encoder {
            rs.reconstruct_data(shards)
                .map_err(|e| io::Error::other(format!("Reed-Solomon reconstruct failed: {e:?}")))
        } else {
            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)
    }

    pub fn verify(&self, shards: &[&[u8]]) -> io::Result<bool> {
        if shards.iter().all(|shard| shard.is_empty()) {
            return Ok(true);
        }

        if let Some(ref rs) = self.encoder {
            rs.verify(shards)
                .map_err(|e| io::Error::other(format!("Reed-Solomon verify failed: {e:?}")))
        } else {
            Ok(true)
        }
    }

    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]);
        }
    }

    if shard_len == 0 {
        for (index, shard) in shards.iter().enumerate() {
            let shard = shard
                .as_ref()
                .ok_or_else(|| io::Error::other(format!("missing shard {index} after data reconstruction")))?;
            if !shard.is_empty() {
                return Err(io::Error::other(format!(
                    "inconsistent shard length at index {index}: got {}, expected {}",
                    shard.len(),
                    shard_len
                )));
            }
        }
        return Ok(());
    }

    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)
}

fn recover_empty_payload_data_shards(
    shards: &mut [Option<Vec<u8>>],
    data_shards: usize,
    parity_shards: usize,
) -> io::Result<bool> {
    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 mut has_present_shard = false;
    for shard in shards.iter().filter_map(|shard| shard.as_ref()) {
        has_present_shard = true;
        if !shard.is_empty() {
            return Ok(false);
        }
    }
    if !has_present_shard {
        return Ok(false);
    }

    for shard in shards.iter_mut().take(data_shards) {
        if shard.is_none() {
            *shard = Some(Vec::new());
        }
    }
    Ok(true)
}

/// 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.
///
/// # Example
/// ```ignore
/// use rustfs_ecstore::api::erasure::Erasure;
/// let erasure = Erasure::new(4, 2, 8);
/// let data = b"hello world";
/// let shards = erasure.encode_data(data).expect("operation should succeed");
/// // Simulate loss and recovery...
/// ```
pub struct Erasure {
    pub data_shards: usize,
    pub parity_shards: usize,
    encoder: Option<ReedSolomonEncoder>,
    legacy_encoder: Option<LegacyReedSolomonEncoder>,
    pub block_size: usize,
    uses_legacy: bool,
    _id: Uuid,
}

impl Default for Erasure {
    fn default() -> Self {
        Self {
            data_shards: 0,
            parity_shards: 0,
            encoder: None,
            legacy_encoder: None,
            block_size: 0,
            uses_legacy: false,
            _id: Uuid::nil(),
        }
    }
}

impl Clone for Erasure {
    fn clone(&self) -> Self {
        Self {
            data_shards: self.data_shards,
            parity_shards: self.parity_shards,
            encoder: self.encoder.clone(),
            legacy_encoder: self.legacy_encoder.clone(),
            block_size: self.block_size,
            uses_legacy: self.uses_legacy,
            _id: self._id, // Shared by clones; this field is unused in hot paths.
        }
    }
}

pub fn calc_shard_size(block_size: usize, data_shards: usize) -> usize {
    block_size.div_ceil(data_shards)
}

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.
    ///
    /// # Panics
    /// Panics when the dimensions are invalid or unsupported. RustFS production
    /// paths use [`Self::try_new`] instead.
    pub fn new(data_shards: usize, parity_shards: usize, block_size: usize) -> Self {
        match Self::try_new(data_shards, parity_shards, block_size) {
            Ok(erasure) => erasure,
            Err(error) => panic!("invalid erasure codec configuration: {error}"),
        }
    }

    /// Create a new Erasure instance with legacy format support.
    ///
    /// When `uses_legacy` is true, uses main-branch shard_size formula and reed-solomon-simd
    /// for decode/reconstruct (for reading and healing old-version files).
    ///
    /// # Panics
    /// Panics when the dimensions are invalid or unsupported. RustFS production
    /// paths use [`Self::try_new_with_options`] instead.
    pub fn new_with_options(data_shards: usize, parity_shards: usize, block_size: usize, uses_legacy: bool) -> Self {
        match Self::try_new_with_options(data_shards, parity_shards, block_size, uses_legacy) {
            Ok(erasure) => erasure,
            Err(error) => panic!("invalid erasure codec configuration: {error}"),
        }
    }

    /// Tries to create a modern Erasure codec.
    ///
    /// # Errors
    /// Returns [`ErasureConstructionError`] when dimensions are zero,
    /// overflow the shard count, exceed the modern codec's supported range, or
    /// the underlying encoder rejects the configuration.
    pub fn try_new(data_shards: usize, parity_shards: usize, block_size: usize) -> Result<Self, ErasureConstructionError> {
        Self::try_new_with_options(data_shards, parity_shards, block_size, false)
    }

    /// Tries to create an Erasure codec using the selected format backend.
    ///
    /// # Errors
    /// Returns [`ErasureConstructionError`] when dimensions are zero,
    /// overflow the shard count, are unsupported by the selected codec, or the
    /// selected encoder fails to initialize.
    pub fn try_new_with_options(
        data_shards: usize,
        parity_shards: usize,
        block_size: usize,
        uses_legacy: bool,
    ) -> Result<Self, ErasureConstructionError> {
        if data_shards == 0 {
            return Err(ErasureConstructionError::ZeroDataShards);
        }
        if block_size == 0 {
            return Err(ErasureConstructionError::ZeroBlockSize);
        }

        let total_shards = data_shards
            .checked_add(parity_shards)
            .ok_or(ErasureConstructionError::ShardCountOverflow {
                data_shards,
                parity_shards,
            })?;

        if parity_shards > 0 {
            if uses_legacy
                && (total_shards > reed_solomon_simd::engine::GF_ORDER
                    || !reed_solomon_simd::ReedSolomonEncoder::supports(data_shards, parity_shards))
            {
                return Err(ErasureConstructionError::UnsupportedLegacyShardCount {
                    data_shards,
                    parity_shards,
                });
            }
            if !uses_legacy && total_shards > MODERN_MAX_TOTAL_SHARDS {
                return Err(ErasureConstructionError::UnsupportedModernShardCount {
                    data_shards,
                    parity_shards,
                });
            }
        }

        let encoder = if !uses_legacy && parity_shards > 0 {
            Some(
                ReedSolomonEncoder::try_new_typed(data_shards, parity_shards)
                    .map_err(|source| ErasureConstructionError::ModernEncoder { source })?,
            )
        } else {
            None
        };

        let legacy_encoder = if uses_legacy && parity_shards > 0 {
            Some(
                LegacyReedSolomonEncoder::new(data_shards, parity_shards)
                    .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?,
            )
        } else {
            None
        };

        Ok(Erasure {
            data_shards,
            parity_shards,
            block_size,
            encoder,
            legacy_encoder,
            uses_legacy,
            _id: Uuid::nil(), // Unused in hot paths; avoid CSPRNG syscall
        })
    }

    /// 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 = "debug", skip_all, fields(data_len=data.len()))]
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
        let shard_size_fn = if self.uses_legacy {
            calc_shard_size_legacy
        } else {
            calc_shard_size
        };
        let per_shard_size = shard_size_fn(data.len(), self.data_shards);
        if per_shard_size == 0 {
            return Ok(vec![Bytes::new(); self.total_shard_count()]);
        }
        let need_total_size = per_shard_size * self.total_shard_count();

        let mut data_buffer = BytesMut::with_capacity(need_total_size);
        data_buffer.extend_from_slice(data);
        data_buffer.resize(need_total_size, 0u8);

        {
            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();

            if self.parity_shards > 0 {
                if self.uses_legacy {
                    if let Some(encoder) = self.legacy_encoder.as_ref() {
                        encoder.encode(data_slices)?;
                    } else {
                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
                    }
                } else if let Some(encoder) = self.encoder.as_ref() {
                    encoder.encode(data_slices)?;
                } 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)
    }

    /// Encode owned data, avoiding a copy when the caller already has a heap buffer.
    /// Falls back to copying into a new buffer if zero-copy conversion fails.
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
        let shard_size_fn = if self.uses_legacy {
            calc_shard_size_legacy
        } else {
            calc_shard_size
        };
        let per_shard_size = shard_size_fn(data.len(), self.data_shards);
        if per_shard_size == 0 {
            return Ok(vec![Bytes::new(); self.total_shard_count()]);
        }
        let need_total_size = per_shard_size * self.total_shard_count();

        // Try zero-copy: Vec<u8> -> Bytes -> BytesMut (succeeds when refcount == 1)
        let mut data_buffer = match Bytes::from(data).try_into_mut() {
            Ok(mut bm) => {
                bm.resize(need_total_size, 0u8);
                bm
            }
            Err(b) => {
                // Rare path: refcount != 1, fall back to copy
                let mut bm = BytesMut::with_capacity(need_total_size);
                bm.extend_from_slice(&b);
                bm.resize(need_total_size, 0u8);
                bm
            }
        };

        {
            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();

            if self.parity_shards > 0 {
                if self.uses_legacy {
                    if let Some(encoder) = self.legacy_encoder.as_ref() {
                        encoder.encode(data_slices)?;
                    } else {
                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
                    }
                } else if let Some(encoder) = self.encoder.as_ref() {
                    encoder.encode(data_slices)?;
                } else {
                    warn!("parity_shards > 0, but encoder is None");
                }
            }
        }

        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)
    }

    /// Encode data from an owned `BytesMut` buffer, avoiding the initial copy
    /// from a borrowed slice into a fresh `BytesMut`.
    ///
    /// Capacity contract: when the caller pre-reserves
    /// `shard_size() * total_shard_count()` bytes (the EC-expanded size of a full
    /// block), the `resize(need_total_size)` below stays within capacity for every
    /// `data_len <= block_size` — both shard-size formulas are monotone in
    /// `data_len` — so this function never reallocates the buffer.
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
        let shard_size_fn = if self.uses_legacy {
            calc_shard_size_legacy
        } else {
            calc_shard_size
        };
        let per_shard_size = shard_size_fn(data_len, self.data_shards);
        if per_shard_size == 0 {
            return Ok(vec![Bytes::new(); self.total_shard_count()]);
        }
        let need_total_size = per_shard_size * self.total_shard_count();

        if data_buffer.len() > data_len {
            data_buffer.truncate(data_len);
        }
        data_buffer.resize(need_total_size, 0u8);

        {
            let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();

            if self.parity_shards > 0 {
                if self.uses_legacy {
                    if let Some(encoder) = self.legacy_encoder.as_ref() {
                        encoder.encode(data_slices)?;
                    } else {
                        warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
                    }
                } else if let Some(encoder) = self.encoder.as_ref() {
                    encoder.encode(data_slices)?;
                } else {
                    warn!("parity_shards > 0, but encoder is None");
                }
            }
        }

        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 data shards in-place.
    ///
    /// # Arguments
    /// * `shards` - Mutable slice of optional shard data. Missing shards should be `None`.
    ///
    /// # Returns
    /// Ok if reconstruction succeeds, error otherwise.
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    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.
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    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() {
                    encoder.reconstruct(shards)?;
                } else {
                    warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
                }
            } else if let Some(encoder) = self.encoder.as_ref() {
                encoder.reconstruct(shards)?;
            } else {
                warn!("parity_shards > 0, but encoder is None");
            }
        }

        Ok(())
    }

    pub(crate) fn decode_data_with_reconstruction_verification(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
        let missing_data_source = shards.iter().take(self.data_shards).any(|shard| shard.is_none());
        let available_shards = shards.iter().filter(|shard| shard.is_some()).count();
        let source_parity = if missing_data_source && available_shards > self.data_shards {
            shards
                .iter()
                .enumerate()
                .skip(self.data_shards)
                .filter_map(|(index, shard)| shard.as_ref().map(|shard| (index, shard.clone())))
                .collect::<Vec<_>>()
        } else {
            Vec::new()
        };

        if source_parity.is_empty() {
            return self.decode_data(shards);
        }

        self.decode_data_and_parity(shards)?;
        for (index, source) in source_parity {
            let Some(rebuilt) = shards[index].as_ref() else {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "missing rebuilt parity shard after read verification",
                ));
            };
            if rebuilt != &source {
                warn!(
                    shard_index = index,
                    data_shards = self.data_shards,
                    parity_shards = self.parity_shards,
                    "erasure decode rejected inconsistent read source shards"
                );
                return Err(io::Error::new(io::ErrorKind::InvalidData, "inconsistent read source shards"));
            }
        }

        Ok(())
    }

    pub(crate) fn verify_data_and_parity(&self, shards: &[Option<Vec<u8>>]) -> io::Result<bool> {
        let expected_shards = self.total_shard_count();
        if shards.len() != expected_shards {
            return Err(io::Error::other(format!(
                "invalid shard count: got {}, expected {}",
                shards.len(),
                expected_shards
            )));
        }
        if self.parity_shards == 0 {
            return Ok(true);
        }

        let mut shard_refs = Vec::with_capacity(expected_shards);
        let mut shard_len = None;
        for (index, shard) in shards.iter().enumerate() {
            let shard = shard
                .as_deref()
                .ok_or_else(|| io::Error::other(format!("missing shard {index} for data/parity verification")))?;
            if let Some(expected_len) = shard_len {
                if shard.len() != expected_len {
                    return Err(io::Error::other(format!(
                        "inconsistent shard length at index {index}: got {}, expected {}",
                        shard.len(),
                        expected_len
                    )));
                }
            } else {
                shard_len = Some(shard.len());
            }
            shard_refs.push(shard);
        }

        if self.uses_legacy {
            if let Some(encoder) = self.legacy_encoder.as_ref() {
                encoder.verify(&shard_refs)
            } else {
                Err(io::Error::other("parity_shards > 0, uses_legacy but legacy_encoder is None"))
            }
        } else if let Some(encoder) = self.encoder.as_ref() {
            encoder.verify(&shard_refs)
        } else {
            Err(io::Error::other("parity_shards > 0, but encoder is None"))
        }
    }

    /// Get the total number of shards (data + parity).
    pub fn total_shard_count(&self) -> usize {
        self.data_shards + self.parity_shards
    }

    /// Whether the erasure dimensions are safe for the shard/offset arithmetic.
    ///
    /// `block_size` and `data_shards` come straight from on-disk metadata; a
    /// corrupted or crafted `xl.meta` can carry zero values that would panic the
    /// `block_size`/`data_shards` divisions in [`Self::shard_size`],
    /// [`Self::shard_file_size`] and [`Self::shard_file_offset`]. Read paths must
    /// reject such metadata before performing those divisions.
    pub fn has_valid_dimensions(&self) -> bool {
        self.block_size > 0 && self.data_shards > 0
    }
    // /// 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 {
        if self.uses_legacy {
            calc_shard_size_legacy(self.block_size, self.data_shards)
        } else {
            calc_shard_size(self.block_size, 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: i64) -> i64 {
        if total_length == 0 {
            return 0;
        }
        if total_length < 0 {
            return total_length;
        }

        let total_length = total_length as usize;
        let shard_size_fn = if self.uses_legacy {
            calc_shard_size_legacy
        } else {
            calc_shard_size
        };

        let num_shards = total_length / self.block_size;
        let last_block_size = total_length % self.block_size;
        let last_shard_size = shard_size_fn(last_block_size, self.data_shards);
        (num_shards * self.shard_size() + last_shard_size) as i64
    }

    /// 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 as i64) as usize;
        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 reader in blocks, calling an async callback for each encoded block.
    /// This method is async and returns the total bytes read after all blocks are processed.
    ///
    /// # Arguments
    /// * `reader` - An async reader implementing AsyncRead + Send + Sync + Unpin
    /// * `mut on_block` - Async callback that receives encoded blocks and returns a Result
    /// * `F` - Callback type: FnMut(Result<Vec<Bytes>, std::io::Error>) -> Future<Output=Result<(), E>> + Send
    /// * `Fut` - Future type returned by the callback
    /// * `E` - Error type returned by the callback
    /// * `R` - Reader type implementing AsyncRead + Send + Sync + Unpin
    ///
    /// # Returns
    /// Result<usize, E> containing total bytes read, or error from callback
    ///
    /// # Errors
    /// Returns error if reading from reader fails or if callback returns error
    pub(crate) 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: AsyncRead + Send + Sync + Unpin,
        F: FnMut(io::Result<Vec<Bytes>>) -> Fut + Send,
        Fut: Future<Output = Result<(), E>> + Send,
    {
        if self.block_size == 0 {
            on_block(Err(io::Error::new(io::ErrorKind::InvalidInput, "erasure block_size must be non-zero"))).await?;
            return Ok(0);
        }

        let block_size = self.block_size;
        let mut total = 0;
        let mut buf = vec![0u8; block_size];
        loop {
            match rustfs_utils::read_full_or_eof(&mut *reader, &mut buf).await {
                Ok(Some(n)) => {
                    debug_assert!(n > 0, "non-zero block_size prevents zero-length reads");
                    warn!("encode_stream_callback_async read n={}", n);
                    total += n;
                    let erasure = self.clone();
                    let encode_buf = std::mem::take(&mut buf);
                    let (res, returned_buf) = match tokio::task::spawn_blocking(move || {
                        let res = erasure.encode_data(&encode_buf[..n]);
                        (res, encode_buf)
                    })
                    .await
                    {
                        Ok(result) => result,
                        Err(err) => {
                            on_block(Err(io::Error::other(format!("EC encode task failed: {err}")))).await?;
                            break;
                        }
                    };
                    buf = returned_buf;
                    on_block(res).await?
                }
                Ok(None) => {
                    warn!("encode_stream_callback_async read unexpected ok");
                    break;
                }
                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                    warn!("encode_stream_callback_async read unexpected eof");
                    break;
                }
                Err(e) => {
                    warn!("encode_stream_callback_async read error={:?}", e);
                    on_block(Err(e)).await?;
                    break;
                }
            }
        }
        Ok(total)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use proptest::collection::{btree_set, vec};
    use proptest::prelude::*;
    use std::pin::Pin;
    use std::task::{Context, Poll};
    use tokio::io::ReadBuf;

    fn erasure_with_invalid_dimensions(data_shards: usize, parity_shards: usize, block_size: usize) -> Erasure {
        let mut erasure = Erasure::default();
        erasure.data_shards = data_shards;
        erasure.parity_shards = parity_shards;
        erasure.block_size = block_size;
        erasure
    }

    fn optional_shards(shards: &[Bytes]) -> Vec<Option<Vec<u8>>> {
        shards.iter().map(|shard| Some(shard.to_vec())).collect()
    }

    fn recover_data(shards: &[Option<Vec<u8>>], data_shards: usize, original_len: usize) -> Vec<u8> {
        let mut recovered = Vec::new();
        for shard in shards.iter().take(data_shards) {
            recovered.extend_from_slice(
                shard
                    .as_ref()
                    .expect("reconstructed data shard should be present after decode_data_and_parity"),
            );
        }
        recovered.truncate(original_len);
        recovered
    }

    fn erasure_recoverability_case_strategy()
    -> impl Strategy<Value = (usize, usize, usize, bool, Vec<u8>, std::collections::BTreeSet<usize>)> {
        (2usize..=8, 1usize..=4, prop::sample::select(vec![64usize, 256, 1024]), any::<bool>()).prop_flat_map(
            |(data_shards, parity_shards, block_size, uses_legacy)| {
                let total_shards = data_shards + parity_shards;
                (
                    Just(data_shards),
                    Just(parity_shards),
                    Just(block_size),
                    Just(uses_legacy),
                    vec(any::<u8>(), 0..=4096),
                    btree_set(0usize..total_shards, 0..=parity_shards),
                )
            },
        )
    }

    fn assert_owned_encode_matches_borrowed(erasure: &Erasure, data: Vec<u8>) {
        let borrowed = erasure.encode_data(&data).expect("borrowed encode should succeed");
        let owned = erasure.encode_data_owned(data).expect("owned encode should succeed");

        assert_eq!(owned, borrowed);
    }

    struct ErrorAfterPartialReader {
        emitted: bool,
    }

    impl AsyncRead for ErrorAfterPartialReader {
        fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
            if !self.emitted {
                self.emitted = true;
                buf.put_slice(&[1]);
                return Poll::Ready(Ok(()));
            }
            Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "partial read failure")))
        }
    }

    struct ImmediateErrorReader;

    impl AsyncRead for ImmediateErrorReader {
        fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Err(io::Error::other("immediate read failure")))
        }
    }

    #[test]
    fn try_new_rejects_zero_and_overflowing_dimensions_with_typed_errors() {
        let Err(zero_data) = Erasure::try_new(0, 2, 64) else {
            panic!("zero data shards must be rejected");
        };
        assert!(matches!(zero_data, ErasureConstructionError::ZeroDataShards));

        let Err(zero_block) = Erasure::try_new(2, 2, 0) else {
            panic!("zero block size must be rejected");
        };
        assert!(matches!(zero_block, ErasureConstructionError::ZeroBlockSize));

        let Err(overflow) = Erasure::try_new(usize::MAX, 1, 64) else {
            panic!("overflowing shard counts must be rejected");
        };
        assert!(matches!(
            overflow,
            ErasureConstructionError::ShardCountOverflow {
                data_shards: usize::MAX,
                parity_shards: 1,
            }
        ));
    }

    #[test]
    fn try_new_rejects_backend_specific_unsupported_dimensions() {
        let Err(modern) = Erasure::try_new(256, 1, 64) else {
            panic!("modern dimensions beyond GF(2^8) must be rejected");
        };
        assert!(matches!(
            modern,
            ErasureConstructionError::UnsupportedModernShardCount {
                data_shards: 256,
                parity_shards: 1,
            }
        ));

        let Err(legacy) = Erasure::try_new_with_options(60_000, 5_000, 64, true) else {
            panic!("unsupported legacy SIMD dimensions must be rejected");
        };
        assert!(matches!(
            legacy,
            ErasureConstructionError::UnsupportedLegacyShardCount {
                data_shards: 60_000,
                parity_shards: 5_000,
            }
        ));
    }

    #[test]
    fn try_new_accepts_exact_backend_shard_limits() {
        let modern_data_shards = MODERN_MAX_TOTAL_SHARDS - 1;
        let modern =
            Erasure::try_new(modern_data_shards, 1, 64).expect("the modern codec's exact field-size boundary must remain valid");
        assert_eq!(modern.total_shard_count(), MODERN_MAX_TOTAL_SHARDS);
        assert!(modern.encoder.is_some());
        assert!(modern.legacy_encoder.is_none());

        let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER / 2;
        let legacy_parity_shards = reed_solomon_simd::engine::GF_ORDER - legacy_data_shards;
        let legacy = Erasure::try_new_with_options(legacy_data_shards, legacy_parity_shards, 64, true)
            .expect("the legacy codec's exact field-size boundary must remain valid");
        assert_eq!(legacy.total_shard_count(), reed_solomon_simd::engine::GF_ORDER);
        assert!(legacy.encoder.is_none());
        assert!(legacy.legacy_encoder.is_some());
    }

    #[test]
    fn try_new_accepts_zero_parity_without_backend_limits() {
        let no_parity = Erasure::try_new(2, 0, 64).expect("zero parity is a valid generic codec configuration");
        assert_eq!(no_parity.total_shard_count(), 2);
        assert!(no_parity.encoder.is_none());
        assert!(no_parity.legacy_encoder.is_none());

        let large_modern = Erasure::try_new(257, 0, 64).expect("zero parity must not inherit the modern backend shard limit");
        assert_eq!(large_modern.total_shard_count(), 257);
        assert!(large_modern.encoder.is_none());
        assert!(large_modern.legacy_encoder.is_none());

        let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER + 1;
        let large_legacy = Erasure::try_new_with_options(legacy_data_shards, 0, 64, true)
            .expect("zero parity must not inherit the legacy backend shard limit");
        assert_eq!(large_legacy.total_shard_count(), legacy_data_shards);
        assert!(large_legacy.encoder.is_none());
        assert!(large_legacy.legacy_encoder.is_none());
    }

    #[test]
    fn try_new_accepts_generic_layouts_above_storage_drive_limits() {
        let erasure = Erasure::try_new(2, 3, 64).expect("generic 2+3 erasure coding must remain supported");
        assert_eq!(erasure.total_shard_count(), 5);
        assert!(erasure.encoder.is_some());
        assert!(erasure.legacy_encoder.is_none());

        let modern = Erasure::try_new(9, 8, 64).expect("the codec must not impose the storage layer's drive limit");
        assert_eq!(modern.total_shard_count(), 17);
        assert!(modern.encoder.is_some());
        assert!(modern.legacy_encoder.is_none());

        let legacy = Erasure::try_new_with_options(9, 8, 64, true)
            .expect("the legacy codec must not impose the storage layer's drive limit");
        assert_eq!(legacy.total_shard_count(), 17);
        assert!(legacy.encoder.is_none());
        assert!(legacy.legacy_encoder.is_some());
    }

    #[test]
    fn try_new_with_options_initializes_only_the_selected_backend() {
        let modern = Erasure::try_new_with_options(4, 2, 64, false).expect("modern codec should construct");
        assert!(modern.encoder.is_some());
        assert!(modern.legacy_encoder.is_none());

        let legacy = Erasure::try_new_with_options(4, 2, 64, true).expect("legacy codec should construct");
        assert!(legacy.encoder.is_none());
        assert!(legacy.legacy_encoder.is_some());
    }

    #[test]
    fn construction_errors_preserve_encoder_sources() {
        let modern = ErasureConstructionError::ModernEncoder {
            source: reed_solomon_erasure::Error::TooManyShards,
        };
        assert!(
            std::error::Error::source(&modern)
                .and_then(|source| source.downcast_ref::<reed_solomon_erasure::Error>())
                .is_some()
        );

        let legacy = ErasureConstructionError::LegacyEncoder {
            source: io::Error::other("legacy encoder failure"),
        };
        assert!(
            std::error::Error::source(&legacy)
                .and_then(|source| source.downcast_ref::<io::Error>())
                .is_some()
        );
    }

    #[test]
    #[should_panic(expected = "invalid erasure codec configuration")]
    fn new_panics_on_invalid_dimensions() {
        let _ = Erasure::new(0, 1, 64);
    }

    #[test]
    #[should_panic(expected = "invalid erasure codec configuration")]
    fn new_with_options_panics_on_invalid_dimensions() {
        let _ = Erasure::new_with_options(1, 1, 0, true);
    }

    #[test]
    fn has_valid_dimensions_rejects_zero_block_size_or_data_shards() {
        // Well-formed erasure metadata is accepted.
        assert!(Erasure::new(4, 2, 64).has_valid_dimensions());
        assert!(Erasure::new_with_options(6, 4, 1, true).has_valid_dimensions());

        // Corrupted on-disk metadata with a zero block_size or zero data_shards
        // must be rejected before it reaches the shard/offset divisions.
        assert!(!erasure_with_invalid_dimensions(4, 2, 0).has_valid_dimensions());
        assert!(!erasure_with_invalid_dimensions(0, 0, 64).has_valid_dimensions());
        assert!(!erasure_with_invalid_dimensions(0, 0, 0).has_valid_dimensions());
    }

    #[tokio::test]
    async fn encode_stream_callback_async_stops_on_reader_errors() {
        let erasure = std::sync::Arc::new(Erasure::new(2, 1, 4));
        let mut partial_error = ErrorAfterPartialReader { emitted: false };
        let mut partial_callbacks = Vec::new();
        let total = erasure
            .clone()
            .encode_stream_callback_async(&mut partial_error, |result| {
                partial_callbacks.push(result.map(|blocks| blocks.len()).map_err(|err| err.kind()));
                async { Ok::<(), io::Error>(()) }
            })
            .await
            .expect("partial read error should not make callback fail");
        assert_eq!(total, 0);
        assert!(
            partial_callbacks.is_empty(),
            "unexpected EOF after a partial read should stop without emitting a block"
        );

        let mut immediate_error = ImmediateErrorReader;
        let mut immediate_callbacks = Vec::new();
        let total = erasure
            .encode_stream_callback_async(&mut immediate_error, |result| {
                immediate_callbacks.push(result.map(|blocks| blocks.len()).map_err(|err| err.kind()));
                async { Ok::<(), io::Error>(()) }
            })
            .await
            .expect("immediate read error should be delivered to callback");
        assert_eq!(total, 0);
        assert_eq!(immediate_callbacks, vec![Err(io::ErrorKind::Other)]);
    }

    #[test]
    fn default_and_legacy_clone_preserve_safe_zero_state_and_restore_data() {
        let default = Erasure::default();
        assert_eq!(default.total_shard_count(), 0);
        assert!(!default.has_valid_dimensions());
        assert_eq!(default.shard_file_size(0), 0);
        assert_eq!(default.shard_file_size(-7), -7);

        let legacy = Erasure::new_with_options(2, 2, 64, true);
        let cloned = legacy.clone();
        assert_eq!(cloned.data_shards, legacy.data_shards);
        assert_eq!(cloned.parity_shards, legacy.parity_shards);
        assert_eq!(cloned.block_size, legacy.block_size);
        assert!(cloned.uses_legacy);

        let data = b"legacy clone should keep independent SIMD caches";
        let encoded = cloned.encode_data(data).expect("legacy clone should encode");
        let mut shards = optional_shards(&encoded);
        shards[0] = None;
        cloned.decode_data(&mut shards).expect("legacy clone should decode");
        assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data);
    }

    #[test]
    fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() {
        let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct");
        let wrong_count: [&[u8]; 1] = [&[]];
        let err = legacy.verify(&wrong_count).expect_err("wrong shard count must be rejected");
        assert!(err.to_string().contains("invalid shard count"));

        let empty: [&[u8]; 4] = [&[], &[], &[], &[]];
        assert!(
            legacy
                .verify(&empty)
                .expect("all-empty legacy shards are internally consistent")
        );

        let erasure = Erasure::new_with_options(2, 2, 64, true);
        let encoded = erasure
            .encode_data(b"legacy verify should compare regenerated parity")
            .expect("legacy encode should succeed");
        let refs = encoded.iter().map(Bytes::as_ref).collect::<Vec<_>>();
        assert!(legacy.verify(&refs).expect("valid legacy shards should verify"));

        let mut corrupt = encoded.iter().map(|shard| shard.to_vec()).collect::<Vec<_>>();
        corrupt[erasure.data_shards][0] ^= 0x80;
        let corrupt_refs = corrupt.iter().map(Vec::as_slice).collect::<Vec<_>>();
        assert!(!legacy.verify(&corrupt_refs).expect("corrupt parity should be detected"));
    }

    #[test]
    fn parity_helper_and_empty_payload_recovery_fail_closed_on_malformed_shards() {
        let mut wrong_count = vec![Some(vec![1])];
        let err = encode_parity_shards(&mut wrong_count, 2, 1, |_| panic!("wrong count must fail before encode"))
            .expect_err("wrong shard count must be rejected");
        assert!(err.to_string().contains("invalid shard count"));

        let mut inconsistent_empty = vec![Some(Vec::new()), Some(vec![1])];
        let err = encode_parity_shards(&mut inconsistent_empty, 1, 1, |_| panic!("zero-length mismatch must fail before encode"))
            .expect_err("mixed empty and non-empty shards must be rejected");
        assert!(err.to_string().contains("inconsistent shard length"));

        let mut inconsistent_non_empty = vec![Some(vec![1]), Some(vec![2, 3])];
        let err = encode_parity_shards(&mut inconsistent_non_empty, 1, 1, |_| {
            panic!("non-empty length mismatch must fail before encode")
        })
        .expect_err("mismatched non-empty shards must be rejected");
        assert!(err.to_string().contains("inconsistent shard length"));

        let mut no_present_empty_payload = vec![None, None, None];
        assert!(
            !recover_empty_payload_data_shards(&mut no_present_empty_payload, 2, 1)
                .expect("all-missing empty payload marker should not be synthesized")
        );
    }

    #[test]
    fn verify_data_and_parity_handles_zero_parity_and_rejects_wrong_count() {
        let no_parity = Erasure::new(2, 0, 64);
        assert!(
            no_parity
                .verify_data_and_parity(&[Some(vec![1]), Some(vec![2, 3])])
                .expect("zero parity requires no parity verification")
        );

        let erasure = Erasure::new(2, 2, 64);
        let err = erasure
            .verify_data_and_parity(&[Some(Vec::new())])
            .expect_err("wrong shard count must fail verification");
        assert!(err.to_string().contains("invalid shard count"));
    }

    #[test]
    fn encode_data_owned_matches_borrowed_path() {
        for uses_legacy in [false, true] {
            let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);

            assert_owned_encode_matches_borrowed(&erasure, Vec::new());
            assert_owned_encode_matches_borrowed(&erasure, b"small payload".to_vec());
            assert_owned_encode_matches_borrowed(&erasure, (0_u8..37).collect());
        }
    }

    #[test]
    fn encode_data_bytes_mut_matches_borrowed_path() {
        for uses_legacy in [false, true] {
            let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);
            let block_size = erasure.block_size;
            let expanded_block_bytes = erasure.shard_size() * erasure.total_shard_count();
            let cases: Vec<Vec<u8>> = vec![
                Vec::new(),
                b"small payload".to_vec(),
                (0_u8..37).collect(),
                vec![0xA5; block_size - 1], // last block one byte short of full
                vec![0x5A; block_size],     // exactly one full block
            ];
            for data in cases {
                let borrowed = erasure.encode_data(&data).expect("borrowed encode should succeed");

                let owned = erasure
                    .encode_data_bytes_mut(BytesMut::from(&data[..]), data.len())
                    .expect("bytesmut encode should succeed");
                assert_eq!(owned, borrowed);

                // Ingest-shaped buffer: spare capacity pre-reserved for the EC-expanded
                // block, exactly what the HP-10 streaming ingest path hands over.
                let mut ingest = BytesMut::with_capacity(expanded_block_bytes.max(block_size));
                ingest.extend_from_slice(&data);
                let preallocated = erasure
                    .encode_data_bytes_mut(ingest, data.len())
                    .expect("preallocated bytesmut encode should succeed");
                assert_eq!(preallocated, borrowed);

                // Buffer longer than data_len (stale bytes past the logical block)
                // must be truncated before padding.
                let mut oversized = BytesMut::from(&data[..]);
                oversized.extend_from_slice(&[0xFF; 8]);
                let truncated = erasure
                    .encode_data_bytes_mut(oversized, data.len())
                    .expect("oversized bytesmut encode should succeed");
                assert_eq!(truncated, borrowed);
            }
        }
    }

    /// HP-10 capacity invariant: both shard-size formulas are monotone in `data_len`,
    /// so pre-reserving `shard_size(block_size) * total_shard_count` covers the
    /// `need_total_size` of every block-or-smaller payload and the ingest buffer
    /// never reallocates inside `encode_data_bytes_mut`.
    #[test]
    fn shard_size_monotonicity_bounds_expanded_block_capacity() {
        for shard_fn in [calc_shard_size, calc_shard_size_legacy] {
            for data_shards in 1usize..=16 {
                for block_size in [1usize, 2, 63, 64, 65, 1024, 4096] {
                    let full = shard_fn(block_size, data_shards);
                    let mut prev = shard_fn(0, data_shards);
                    for data_len in 1..=block_size {
                        let cur = shard_fn(data_len, data_shards);
                        assert!(cur >= prev, "shard size must be monotone (len {data_len}, shards {data_shards})");
                        assert!(cur <= full, "per-block shard size must not exceed the full-block bound");
                        prev = cur;
                    }
                }
                // Spot-check the production-scale block size at its boundaries.
                let block_size = 1usize << 20;
                let full = shard_fn(block_size, data_shards);
                for data_len in [0usize, 1, block_size / 2, block_size - 1, block_size] {
                    assert!(shard_fn(data_len, data_shards) <= full);
                }
            }
        }
    }

    #[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_rejects_missing_data_above_parity_budget() {
        for uses_legacy in [false, true] {
            let erasure = Erasure::new_with_options(3, 2, 64, uses_legacy);
            let data = b"read restore must fail when too many data shards are gone";
            let encoded = erasure.encode_data(data).expect("encode should succeed");
            let mut shards = optional_shards(&encoded);
            shards[0] = None;
            shards[1] = None;
            shards[2] = None;

            let err = erasure
                .decode_data(&mut shards)
                .expect_err("decode_data must fail when missing data shards exceed parity");

            assert!(!err.to_string().is_empty());
            assert!(shards.iter().take(erasure.data_shards).any(Option::is_none));
        }
    }

    #[test]
    fn decode_data_and_parity_rejects_wrong_shard_count() {
        for uses_legacy in [false, true] {
            let erasure = Erasure::new_with_options(4, 2, 128, uses_legacy);
            let data = b"wrong shard count should not be accepted as a restored object";
            let encoded = erasure.encode_data(data).expect("encode should succeed");
            let mut shards = optional_shards(&encoded);
            let _ = shards.pop();

            let err = erasure
                .decode_data_and_parity(&mut shards)
                .expect_err("decode_data_and_parity must reject wrong shard count");

            assert!(!err.to_string().is_empty());
        }
    }

    #[test]
    fn decode_data_with_verification_rejects_corrupt_surplus_parity() {
        let erasure = Erasure::new(3, 2, 128);
        let data = b"verified reads must fail closed on stale surplus parity";
        let encoded = erasure.encode_data(data).expect("encode should succeed");
        let mut shards = optional_shards(&encoded);
        shards[1] = None;
        shards[erasure.data_shards].as_mut().expect("parity shard should be present")[0] ^= 0x7d;

        let err = erasure
            .decode_data_with_reconstruction_verification(&mut shards)
            .expect_err("verified decode must reject inconsistent parity");

        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
    }

    #[test]
    fn verify_data_and_parity_rejects_missing_and_mismatched_shards() {
        let erasure = Erasure::new(4, 2, 128);
        let data = b"verification must reject incomplete or malformed shard sets";
        let encoded = erasure.encode_data(data).expect("encode should succeed");
        let mut shards = optional_shards(&encoded);

        shards[0] = None;
        let missing = erasure
            .verify_data_and_parity(&shards)
            .expect_err("verification must reject missing shards");
        assert!(missing.to_string().contains("missing shard"));

        let mut mismatched = optional_shards(&encoded);
        let _ = mismatched[0].as_mut().expect("data shard should be present").pop();
        let mismatch = erasure
            .verify_data_and_parity(&mismatched)
            .expect_err("verification must reject mismatched shard lengths");
        assert!(mismatch.to_string().contains("inconsistent shard length"));
    }

    #[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 legacy_decode_data_and_parity_reconstructs_empty_object_shards() {
        let erasure = Erasure::new_with_options(3, 3, 64, true);
        let encoded = erasure.encode_data(&[]).expect("empty encode should succeed");
        let mut shards = optional_shards(&encoded);
        shards[1] = None;
        shards[4] = None;

        erasure
            .decode_data_and_parity(&mut shards)
            .expect("empty decode should rebuild missing shards without SIMD");

        for (index, shard) in shards.iter().enumerate() {
            assert_eq!(
                shard.as_deref(),
                Some(encoded[index].as_ref()),
                "empty shard {index} should match encoded source"
            );
        }
    }

    #[test]
    fn test_shard_file_size_cases2() {
        let erasure = Erasure::new(12, 4, 1024 * 1024);
        assert_eq!(erasure.shard_file_size(1572864), 131073);
    }

    proptest! {
        #[test]
        fn decode_data_and_parity_round_trips_bounded_recoverability(
            (data_shards, parity_shards, block_size, uses_legacy, data, missing_indices) in
                erasure_recoverability_case_strategy(),
        ) {
            let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, uses_legacy);
            let encoded = erasure.encode_data(&data)
                .expect("encode_data should succeed for bounded recoverability property cases");
            let mut shards = optional_shards(&encoded);

            for index in &missing_indices {
                shards[*index] = None;
            }

            erasure.decode_data_and_parity(&mut shards)
                .expect("decode_data_and_parity should recover when missing shard count does not exceed parity");

            let recovered = recover_data(&shards, data_shards, data.len());
            prop_assert_eq!(recovered, data);

            for (index, shard) in shards.iter().enumerate() {
                prop_assert_eq!(
                    shard.as_deref(),
                    Some(encoded[index].as_ref()),
                    "reconstructed shard {} should match the original encoded shard",
                    index
                );
            }
        }
    }

    #[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

        // MinIO-compatible: 1248739/8=156092, last=3, ceil(3/4)=1, 156092*2+1=312185
        assert_eq!(erasure.shard_file_size(1248739), 312185);

        // MinIO-compatible: 43/8=5, last=3, ceil(3/4)=1, 5*2+1=11
        assert_eq!(erasure.shard_file_size(43), 11);

        // 1572864 with block_size=8: 196608 full blocks, last=0, 196608*2+0=393216
        assert_eq!(erasure.shard_file_size(1572864), 393216);
    }

    #[test]
    fn test_encode_decode_roundtrip() {
        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 1024; // SIMD mode
        let erasure = Erasure::new(data_shards, parity_shards, block_size);

        // Use sufficient test data for SIMD optimization
        let test_data = b"SIMD mode test data for encoding and decoding roundtrip verification with sufficient length to ensure shard size requirements are met for proper SIMD optimization.".repeat(20); // ~3KB for SIMD

        let data = &test_data;
        let encoded_shards = erasure.encode_data(data).expect("operation should succeed");
        assert_eq!(encoded_shards.len(), data_shards + parity_shards);

        // Create decode input with some shards missing, convert to the format expected by decode_data
        let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
        for i in 0..data_shards {
            decode_input[i] = Some(encoded_shards[i].to_vec());
        }

        erasure.decode_data(&mut decode_input).expect("operation should succeed");

        // Recover original data
        let mut recovered = Vec::new();
        for shard in decode_input.iter().take(data_shards) {
            recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
        }
        recovered.truncate(data.len());
        assert_eq!(&recovered, data);
    }

    #[test]
    fn test_encode_decode_large_1m() {
        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 512 * 3; // SIMD mode
        let erasure = Erasure::new(data_shards, parity_shards, block_size);

        // Generate 1MB test data
        let data: Vec<u8> = (0..1048576).map(|i| (i % 256) as u8).collect();

        let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
        assert_eq!(encoded_shards.len(), data_shards + parity_shards);

        // Create decode input with some shards missing, convert to the format expected by decode_data
        let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
        for i in 0..data_shards {
            decode_input[i] = Some(encoded_shards[i].to_vec());
        }

        erasure.decode_data(&mut decode_input).expect("operation should succeed");

        // Recover original data
        let mut recovered = Vec::new();
        for shard in decode_input.iter().take(data_shards) {
            recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
        }
        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).expect("operation should succeed");
        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_legacy_shard_size_and_file_size() {
        let erasure = Erasure::new_with_options(4, 2, 8, true);
        assert_eq!(erasure.shard_size(), 2);
        assert_eq!(calc_shard_size_legacy(8, 4), 2);
        assert_eq!(calc_shard_size_legacy(1, 4), 2);
        assert_eq!(erasure.shard_file_size(33), 10);
        assert_eq!(erasure.shard_file_size(0), 0);
    }

    #[test]
    fn test_legacy_encode_decode_roundtrip() {
        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 1024;
        let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, true);

        let data = b"Legacy encode/decode roundtrip test data with sufficient length.".repeat(20);
        let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
        assert_eq!(encoded_shards.len(), data_shards + parity_shards);

        let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
        for i in 0..data_shards {
            decode_input[i] = Some(encoded_shards[i].to_vec());
        }

        erasure.decode_data(&mut decode_input).expect("operation should succeed");

        let mut recovered = Vec::new();
        for shard in decode_input.iter().take(data_shards) {
            recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
        }
        recovered.truncate(data.len());
        assert_eq!(&recovered, &data);
    }

    #[test]
    fn test_legacy_decode_with_missing_shards() {
        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 256;
        let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, true);

        let data = b"Legacy decode with missing shards test.".repeat(10);
        let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");

        let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|s| Some(s.to_vec())).collect();
        shards_opt[1] = None;
        shards_opt[5] = None;

        erasure.decode_data(&mut shards_opt).expect("operation should succeed");

        let mut recovered = Vec::new();
        for shard in shards_opt.iter().take(data_shards) {
            recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
        }
        recovered.truncate(data.len());
        assert_eq!(&recovered, &data);
    }

    #[test]
    fn test_shard_file_offset() {
        let erasure = Erasure::new(8, 8, 1024 * 1024);
        let offset = erasure.shard_file_offset(0, 86, 86);
        println!("offset={offset}");
        assert!(offset > 0);

        let total_length = erasure.shard_file_size(86);
        println!("total_length={total_length}");
        assert!(total_length > 0);
    }

    #[tokio::test]
    async fn test_encode_stream_callback_async_error_propagation() {
        use std::io::Cursor;
        use std::sync::Arc;
        use tokio::sync::mpsc;

        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 1024; // SIMD mode
        let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));

        // Use test data suitable for SIMD mode
        let data =
            b"Async error test data with sufficient length to meet requirements for proper testing and validation.".repeat(20); // ~2KB

        let mut reader = Cursor::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 reader, move |res| {
                    let tx = tx.clone();
                    async move {
                        let shards = res.expect("operation should succeed");
                        tx.send(shards).await.expect("operation should succeed");
                        Ok(())
                    }
                })
                .await
                .expect("operation should succeed");
        });
        let result = handle.await;
        assert!(result.is_ok());
        let collected_shards = rx.recv().await.expect("operation should succeed");
        assert_eq!(collected_shards.len(), data_shards + parity_shards);
    }

    #[tokio::test]
    async fn test_encode_stream_callback_async_channel_decode() {
        use std::io::Cursor;
        use std::sync::Arc;
        use tokio::sync::mpsc;

        let data_shards = 4;
        let parity_shards = 2;
        let block_size = 1024; // SIMD mode
        let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));

        // Use test data that fits in exactly one block to avoid multi-block complexity
        let data =
            b"Channel async callback test data with sufficient length to ensure proper operation and validation requirements."
                .repeat(8); // ~1KB

        let data_clone = data.clone(); // Clone for later comparison
        let mut reader = Cursor::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 reader, move |res| {
                    let tx = tx.clone();
                    async move {
                        let shards = res.expect("operation should succeed");
                        tx.send(shards).await.expect("operation should succeed");
                        Ok(())
                    }
                })
                .await
                .expect("operation should succeed");
        });
        let result = handle.await;
        assert!(result.is_ok());
        let shards = rx.recv().await.expect("operation should succeed");
        assert_eq!(shards.len(), data_shards + parity_shards);

        // Test decode using the old API that operates in-place
        let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
        for i in 0..data_shards {
            decode_input[i] = Some(shards[i].to_vec());
        }
        erasure.decode_data(&mut decode_input).expect("operation should succeed");

        // Recover original data
        let mut recovered = Vec::new();
        for shard in decode_input.iter().take(data_shards) {
            recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
        }
        recovered.truncate(data_clone.len());
        assert_eq!(&recovered, &data_clone);
    }

    #[tokio::test]
    async fn test_encode_stream_callback_async_reports_zero_block_size() {
        use std::io::Cursor;
        use std::sync::{Arc, Mutex};

        let erasure = Arc::new(erasure_with_invalid_dimensions(1, 0, 0));
        let mut reader = Cursor::new(b"payload".to_vec());
        let observed = Arc::new(Mutex::new(None));
        let observed_clone = observed.clone();

        let total = erasure
            .encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
                let observed = observed_clone.clone();
                async move {
                    let err = res.expect_err("zero block size should report an error");
                    *observed.lock().expect("operation should succeed") = Some((err.kind(), err.to_string()));
                    Ok(())
                }
            })
            .await
            .expect("callback should handle the zero block size error");

        assert_eq!(total, 0);
        let observed = observed.lock().expect("operation should succeed");
        let (kind, message) = observed.as_ref().expect("callback should be invoked once");
        assert_eq!(*kind, io::ErrorKind::InvalidInput);
        assert!(message.contains("block_size"));
    }

    // SIMD mode specific tests
    mod simd_tests {
        use super::*;

        #[test]
        fn test_simd_encode_decode_roundtrip() {
            let data_shards = 4;
            let parity_shards = 2;
            let block_size = 1024; // Use larger block size for SIMD mode
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Use data that will create shards >= 512 bytes for SIMD optimization
            let test_data = b"SIMD mode test data for encoding and decoding roundtrip verification with sufficient length to ensure shard size requirements are met for proper SIMD optimization and validation.";
            let data = test_data.repeat(25); // Create much larger data: ~5KB total, ~1.25KB per shard

            let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
            assert_eq!(encoded_shards.len(), data_shards + parity_shards);

            // Create decode input with some shards missing
            let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|shard| Some(shard.to_vec())).collect();

            // Lose one data shard and one parity shard (should still be recoverable)
            shards_opt[1] = None; // Lose second data shard
            shards_opt[5] = None; // Lose second parity shard

            erasure.decode_data(&mut shards_opt).expect("operation should succeed");

            // Verify recovered data
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());
            assert_eq!(&recovered, &data);
        }

        #[test]
        fn test_simd_all_zero_data() {
            let data_shards = 4;
            let parity_shards = 2;
            let block_size = 1024; // Use larger block size for SIMD mode
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Create all-zero data that ensures adequate shard size for SIMD optimization
            let data = vec![0u8; 1024]; // 1KB of zeros, each shard will be 256 bytes

            let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
            assert_eq!(encoded_shards.len(), data_shards + parity_shards);

            // Verify that all data shards are zeros
            for (i, shard) in encoded_shards.iter().enumerate().take(data_shards) {
                assert!(shard.iter().all(|&x| x == 0), "Data shard {i} should be all zeros");
            }

            // Test recovery with some shards missing
            let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|shard| Some(shard.to_vec())).collect();

            // Lose maximum recoverable shards (equal to parity_shards)
            shards_opt[0] = None; // Lose first data shard
            shards_opt[4] = None; // Lose first parity shard

            erasure.decode_data(&mut shards_opt).expect("operation should succeed");

            // Verify recovered data is still all zeros
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());
            assert!(recovered.iter().all(|&x| x == 0), "Recovered data should be all zeros");
        }

        #[test]
        fn test_simd_large_data_1kb() {
            let data_shards = 8;
            let parity_shards = 4;
            let block_size = 1024; // 1KB block size optimal for SIMD
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Create 1KB of test data
            let mut data = Vec::with_capacity(1024);
            for i in 0..1024 {
                data.push((i % 256) as u8);
            }

            let shards = erasure.encode_data(&data).expect("operation should succeed");
            assert_eq!(shards.len(), data_shards + parity_shards);

            // Simulate the loss of multiple shards
            let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
            shards_opt[0] = None;
            shards_opt[3] = None;
            shards_opt[9] = None; // Parity shard
            shards_opt[11] = None; // Parity shard

            // Decode
            erasure.decode_data(&mut shards_opt).expect("operation should succeed");

            // Recover original data
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());
            assert_eq!(&recovered, &data);
        }

        #[test]
        fn test_simd_minimum_shard_size() {
            let data_shards = 4;
            let parity_shards = 2;
            let block_size = 256; // Use 256 bytes to ensure sufficient shard size
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Create data that will result in 64+ byte shards
            let data = vec![0x42u8; 200]; // 200 bytes, should create ~50 byte shards per data shard

            let shards = erasure.encode_data(&data).expect("minimum shard size data should encode");
            println!("SIMD encoding succeeded with shard size: {}", shards[0].len());

            // Test decoding
            let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
            shards_opt[1] = None;

            erasure
                .decode_data(&mut shards_opt)
                .expect("minimum shard size data should decode");
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());
            assert_eq!(&recovered, &data);
        }

        #[test]
        fn test_simd_maximum_erasures() {
            let data_shards = 5;
            let parity_shards = 3;
            let block_size = 512;
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            let data =
                b"Testing maximum erasure capacity with SIMD Reed-Solomon implementation for robustness verification!".repeat(3);

            let shards = erasure.encode_data(&data).expect("operation should succeed");

            // Lose exactly the maximum number of shards (equal to parity_shards)
            let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
            shards_opt[0] = None; // Data shard
            shards_opt[2] = None; // Data shard  
            shards_opt[6] = None; // Parity shard

            // Should succeed with maximum erasures
            erasure.decode_data(&mut shards_opt).expect("operation should succeed");

            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());
            assert_eq!(&recovered, &data);
        }

        /// Generates 7557 bytes identical to MinIO generateCompatTestData.
        fn generate_compat_test_data(size: usize) -> Vec<u8> {
            (0..size).map(|i| ((i * 7 + 13) % 256) as u8).collect()
        }

        /// Verifies reed-solomon-simd produces same shards.
        /// Data shards (0-3) must match for MinIO to read RustFS part files.
        /// Parity shards (4-5) differ: reed-solomon-simd vs klauspost use different RS encoding.
        /// Run: cargo test -p rustfs-ecstore test_reed_solomon_compat
        #[test]
        fn test_reed_solomon_compat() {
            let data = generate_compat_test_data(7557);
            let erasure = Erasure::new(4, 2, 7557);
            let shards = erasure.encode_data(&data).expect("operation should succeed");
            assert_eq!(shards.len(), 6, "expected 6 shards (4 data + 2 parity)");

            // Per-shard HighwayHash
            let expected_hashes: [&str; 6] = [
                "fb3db9338e610cec541504ddae4b0bfd54445bcbd45318cf21f35f024240914d", // data 0
                "a545269a3196e18e77ef9f5ec6e735a4f4ebe82d342db666b11a5256eb305720", // data 1
                "2adbf0058f36c4cbcb5c9c16c38a6530c54198dfe504179a6f92d2349f245318", // data 2
                "898e6d060b0cb4f0e830add7e1f936bc8b78442bf582283ee244a3a058602db8", // data 3
                "4a20460bca044b3a777b26f2b0bcd371e3eab2f156f84778be3ccd8edd521ef2", // parity 4
                "eb8ba4c0db15ca910d58d031f74e4601ba2fed62ad03ec29cadde3367ab0d415", // parity 5
            ];

            let mut data_shards_match = true;
            let mut parity_shards_match = true;
            for (i, shard) in shards.iter().enumerate() {
                let hash = rustfs_utils::HashAlgorithm::HighwayHash256S.hash_encode(shard);
                let got = hex_simd::encode_to_string(hash.as_ref(), hex_simd::AsciiCase::Lower);
                let matches = got == expected_hashes[i];
                if i < 4 {
                    data_shards_match &= matches;
                } else {
                    parity_shards_match &= matches;
                }
                if !matches {
                    eprintln!(
                        "Shard {} ({}): got {} want {}",
                        i,
                        if i < 4 { "data" } else { "parity" },
                        got,
                        expected_hashes[i]
                    );
                }
            }
            assert!(data_shards_match, "Data shards (0-3) must match");
            assert!(parity_shards_match, "Parity shards (4-5): reed-solomon-simd differs");
        }

        #[test]
        fn test_simd_small_data_handling() {
            let data_shards = 4;
            let parity_shards = 2;
            let block_size = 32; // Small block size for testing edge cases
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Use small data to test SIMD handling of small shards
            let small_data = b"tiny!123".to_vec(); // 8 bytes data

            // Test encoding with small data
            let shards = erasure.encode_data(&small_data).expect("small data should encode");
            println!("SIMD encoding succeeded: {} bytes into {} shards", small_data.len(), shards.len());
            assert_eq!(shards.len(), data_shards + parity_shards);

            // Test decoding
            let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|shard| Some(shard.to_vec())).collect();

            // Lose some shards to test recovery
            shards_opt[1] = None; // Lose one data shard
            shards_opt[4] = None; // Lose one parity shard

            erasure.decode_data(&mut shards_opt).expect("small data should decode");
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(small_data.len());
            println!("recovered: {recovered:?}");
            println!("small_data: {small_data:?}");
            assert_eq!(&recovered, &small_data);
        }

        #[test]
        fn test_simd_large_block_1mb() {
            let data_shards = 6;
            let parity_shards = 3;
            let block_size = 1024 * 1024; // 1MB block size
            let erasure = Erasure::new(data_shards, parity_shards, block_size);

            // Build 2 MB of test data so multiple 1 MB chunks are exercised
            let mut data = Vec::with_capacity(2 * 1024 * 1024);
            for i in 0..(2 * 1024 * 1024) {
                data.push((i % 256) as u8);
            }

            println!("🚀 Testing SIMD with 1MB block size and 2MB data");
            println!(
                "📊 Data shards: {}, Parity shards: {}, Total data: {}KB",
                data_shards,
                parity_shards,
                data.len() / 1024
            );

            // Encode the data
            let start = std::time::Instant::now();
            let shards = erasure.encode_data(&data).expect("operation should succeed");
            let encode_duration = start.elapsed();

            println!("⏱️  Encoding completed in: {encode_duration:?}");
            println!("📦 Generated {} shards, each shard size: {}KB", shards.len(), shards[0].len() / 1024);

            assert_eq!(shards.len(), data_shards + parity_shards);

            // Verify that each shard is large enough for SIMD optimization
            for (i, shard) in shards.iter().enumerate() {
                println!("🔍 Shard {}: {} bytes ({}KB)", i, shard.len(), shard.len() / 1024);
                assert!(shard.len() >= 512, "Shard {} is too small for SIMD: {} bytes", i, shard.len());
            }

            // Simulate data loss - lose maximum recoverable number of shards
            let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
            shards_opt[0] = None; // Lose 1st data shard
            shards_opt[2] = None; // Lose 3rd data shard  
            shards_opt[8] = None; // Lose 3rd parity shard (index 6+3-1=8)

            println!("💥 Simulated loss of 3 shards (max recoverable with 3 parity shards)");

            // Decode and recover data
            let start = std::time::Instant::now();
            erasure.decode_data(&mut shards_opt).expect("operation should succeed");
            let decode_duration = start.elapsed();

            println!("⏱️  Decoding completed in: {decode_duration:?}");

            // Verify recovered data integrity
            let mut recovered = Vec::new();
            for shard in shards_opt.iter().take(data_shards) {
                recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
            }
            recovered.truncate(data.len());

            assert_eq!(recovered.len(), data.len());
            assert_eq!(&recovered, &data, "Data mismatch after recovery!");

            println!("✅ Successfully verified data integrity after recovery");
            println!("📈 Performance summary:");
            println!(
                "   - Encode: {:?} ({:.2} MB/s)",
                encode_duration,
                (data.len() as f64 / (1024.0 * 1024.0)) / encode_duration.as_secs_f64()
            );
            println!(
                "   - Decode: {:?} ({:.2} MB/s)",
                decode_duration,
                (data.len() as f64 / (1024.0 * 1024.0)) / decode_duration.as_secs_f64()
            );
        }

        #[tokio::test]
        async fn test_simd_stream_callback() {
            use std::io::Cursor;
            use std::sync::Arc;
            use tokio::sync::mpsc;

            let data_shards = 4;
            let parity_shards = 2;
            let block_size = 256; // Larger block for SIMD
            let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));

            let test_data = b"SIMD stream processing test with sufficient data length for multiple blocks and proper SIMD optimization verification!";
            let data = test_data.repeat(5); // Create owned Vec<u8>
            let data_clone = data.clone(); // Clone for later comparison
            let mut reader = Cursor::new(data);

            let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(16);
            let erasure_clone = erasure.clone();

            let handle = tokio::spawn(async move {
                erasure_clone
                    .encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
                        let tx = tx.clone();
                        async move {
                            let shards = res.expect("operation should succeed");
                            tx.send(shards).await.expect("operation should succeed");
                            Ok(())
                        }
                    })
                    .await
                    .expect("operation should succeed");
            });

            let mut all_blocks = Vec::new();
            while let Some(block) = rx.recv().await {
                all_blocks.push(block);
            }
            handle.await.expect("operation should succeed");

            // Verify we got multiple blocks
            assert!(all_blocks.len() > 1, "Should have multiple blocks for stream test");

            // Test recovery for each 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();
                // Lose one data shard and one parity shard
                shards_opt[1] = None;
                shards_opt[5] = None;

                erasure.decode_data(&mut shards_opt).expect("operation should succeed");

                for shard in shards_opt.iter().take(data_shards) {
                    recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
                }
            }

            recovered.truncate(data_clone.len());
            assert_eq!(&recovered, &data_clone);
        }
    }
}