rustfs_erasure_codec/core/

stream.rs

//! Streaming encode/verify/reconstruct for large data.
//!
//! Provides [`encode_stream`](crate::galois_8::ReedSolomon::encode_stream),
//! [`verify_stream`](crate::galois_8::ReedSolomon::verify_stream), and
//! [`reconstruct_stream`](crate::galois_8::ReedSolomon::reconstruct_stream) methods that
//! process data in configurable blocks, avoiding the need to load entire
//! files into memory.
//!
//! # Example
//!
//! ```no_run
//! use rustfs_erasure_codec::galois_8::ReedSolomon;
//! use rustfs_erasure_codec::stream::StreamOptions;
//! use std::fs::File;
//! use std::io::BufReader;
//!
//! let rs = ReedSolomon::new(10, 4).unwrap();
//!
//! let mut data_readers: Vec<BufReader<File>> = (0..10)
//!     .map(|i| BufReader::new(File::open(format!("data_{i}.bin")).unwrap()))
//!     .collect();
//! let mut parity_writers: Vec<File> = (0..4)
//!     .map(|i| File::create(format!("parity_{i}.bin")).unwrap())
//!     .collect();
//!
//! let opts = StreamOptions::new().with_block_size(4 * 1024 * 1024);
//! rs.encode_stream(&mut data_readers, &mut parity_writers, &opts).unwrap();
//! ```

use std::io::{Error, Read};
use std::mem;

// ---------------------------------------------------------------------------
// StreamOptions
// ---------------------------------------------------------------------------

/// Configuration for streaming encode/verify/reconstruct operations.
#[derive(Debug, Clone)]
pub struct StreamOptions {
    /// Block size in bytes for each read/write cycle. Default: 4 MiB.
    ///
    /// Larger blocks reduce syscall overhead but increase memory usage.
    /// Recommended range: 256 KiB – 16 MiB.
    pub block_size: usize,
}

impl Default for StreamOptions {
    fn default() -> Self {
        Self {
            block_size: 4 * 1024 * 1024,
        }
    }
}

impl StreamOptions {
    /// Create a new `StreamOptions` with default settings.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the block size (minimum 1 KiB).
    pub fn with_block_size(mut self, size: usize) -> Self {
        const MIN_BLOCK_SIZE_BYTES: usize = 1024;
        const MAX_BLOCK_SIZE_BYTES: usize = 16 * 1024 * 1024;
        self.block_size = size.clamp(MIN_BLOCK_SIZE_BYTES, MAX_BLOCK_SIZE_BYTES);
        self
    }
}

// ---------------------------------------------------------------------------
// StreamError
// ---------------------------------------------------------------------------

/// Error returned by streaming operations.
#[derive(Debug)]
pub struct StreamError {
    /// Index of the shard that caused the error.
    pub shard_index: usize,
    /// The kind of error.
    pub kind: StreamErrorKind,
}

/// Classification of [`StreamError`].
#[derive(Debug)]
pub enum StreamErrorKind {
    /// I/O error while reading a shard.
    Read(std::io::Error),
    /// I/O error while writing a shard.
    Write(std::io::Error),
    /// Error from the underlying codec (encode/verify/reconstruct).
    Codec(crate::Error),
}

impl core::fmt::Display for StreamError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match &self.kind {
            StreamErrorKind::Read(e) => {
                write!(f, "read error on shard {}: {}", self.shard_index, e)
            }
            StreamErrorKind::Write(e) => {
                write!(f, "write error on shard {}: {}", self.shard_index, e)
            }
            StreamErrorKind::Codec(e) => {
                write!(f, "codec error on shard {}: {}", self.shard_index, e)
            }
        }
    }
}

impl std::error::Error for StreamError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match &self.kind {
            StreamErrorKind::Read(e) => Some(e),
            StreamErrorKind::Write(e) => Some(e),
            StreamErrorKind::Codec(e) => Some(e),
        }
    }
}

impl StreamError {
    fn read(shard_index: usize, e: std::io::Error) -> Self {
        Self {
            shard_index,
            kind: StreamErrorKind::Read(e),
        }
    }

    fn write(shard_index: usize, e: std::io::Error) -> Self {
        Self {
            shard_index,
            kind: StreamErrorKind::Write(e),
        }
    }

    fn codec(shard_index: usize, e: crate::Error) -> Self {
        Self {
            shard_index,
            kind: StreamErrorKind::Codec(e),
        }
    }
}

fn take_stream_error(
    first_error: &std::sync::Mutex<Option<StreamError>>,
    fallback_message: &'static str,
) -> StreamError {
    match first_error.lock() {
        Ok(mut guard) => guard
            .take()
            .unwrap_or_else(|| StreamError::read(0, Error::other(fallback_message))),
        Err(poisoned) => {
            let mut guard = poisoned.into_inner();
            guard
                .take()
                .unwrap_or_else(|| StreamError::read(0, Error::other(fallback_message)))
        }
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Read up to `max_len` bytes from each reader into the corresponding
/// buffer, retrying on `Interrupted`.  Returns `Ok((all_eof, actual_len))`
/// where `all_eof` is `true` if every reader was already at EOF, and
/// `actual_len` is the number of bytes read (same across all readers,
/// with short reads zero-padded).
fn read_block<R: std::io::Read>(
    readers: &mut [R],
    buffers: &mut [Vec<u8>],
    max_len: usize,
) -> Result<(bool, usize), StreamError> {
    let mut any_data = false;
    let mut actual_len = 0;

    for (i, (reader, buf)) in readers.iter_mut().zip(buffers.iter_mut()).enumerate() {
        buf.resize(max_len, 0);
        let mut total = 0;

        while total < max_len {
            match reader.read(&mut buf[total..]) {
                Ok(0) => break,
                Ok(n) => {
                    total += n;
                    any_data = true;
                }
                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(e) => return Err(StreamError::read(i, e)),
            }
        }

        if total > actual_len {
            actual_len = total;
        }

        // Zero-fill remainder so all buffers have identical length.
        for byte in &mut buf[total..] {
            *byte = 0;
        }
    }

    Ok((!any_data, actual_len))
}

/// Read from all shard readers (data + parity) for verify_stream.
fn read_block_all<R: std::io::Read>(
    readers: &mut [R],
    buffers: &mut [Vec<u8>],
    max_len: usize,
) -> Result<(bool, usize), StreamError> {
    let mut any_data = false;
    let mut actual_len = 0;

    for (i, (reader, buf)) in readers.iter_mut().zip(buffers.iter_mut()).enumerate() {
        buf.resize(max_len, 0);
        let mut total = 0;

        while total < max_len {
            match reader.read(&mut buf[total..]) {
                Ok(0) => break,
                Ok(n) => {
                    total += n;
                    any_data = true;
                }
                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(e) => return Err(StreamError::read(i, e)),
            }
        }

        if total > actual_len {
            actual_len = total;
        }

        for byte in &mut buf[total..] {
            *byte = 0;
        }
    }

    Ok((!any_data, actual_len))
}

/// Write `len` bytes from each buffer to the corresponding writer.
fn write_block<W: std::io::Write>(
    writers: &mut [W],
    buffers: &[Vec<u8>],
    len: usize,
    shard_offset: usize,
) -> Result<(), StreamError> {
    for (i, (writer, buf)) in writers.iter_mut().zip(buffers.iter()).enumerate() {
        let mut written = 0;
        while written < len {
            match writer.write(&buf[written..len]) {
                Ok(0) => {
                    return Err(StreamError::write(
                        shard_offset + i,
                        std::io::Error::new(std::io::ErrorKind::WriteZero, "write returned 0"),
                    ));
                }
                Ok(n) => written += n,
                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                Err(e) => return Err(StreamError::write(shard_offset + i, e)),
            }
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Parallel I/O helpers (rayon)
// ---------------------------------------------------------------------------

/// Parallel version of `read_block` — reads all shards concurrently.
fn read_block_par<R: std::io::Read + Send>(
    readers: &mut [R],
    buffers: &mut [Vec<u8>],
    max_len: usize,
) -> Result<(bool, usize), StreamError> {
    use rayon::prelude::*;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

    let any_data = AtomicBool::new(false);
    let actual_len = AtomicUsize::new(0);
    let first_error: std::sync::Mutex<Option<StreamError>> = std::sync::Mutex::new(None);

    readers
        .par_iter_mut()
        .zip(buffers.par_iter_mut())
        .enumerate()
        .try_for_each(|(i, (reader, buf))| {
            buf.resize(max_len, 0);
            let mut total = 0;

            while total < max_len {
                match reader.read(&mut buf[total..]) {
                    Ok(0) => break,
                    Ok(n) => {
                        total += n;
                        any_data.store(true, Ordering::Relaxed);
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                    Err(e) => {
                        if let Ok(mut guard) = first_error.lock()
                            && guard.is_none()
                        {
                            *guard = Some(StreamError::read(i, e));
                        }
                        return Err(());
                    }
                }
            }

            actual_len.fetch_max(total, Ordering::Relaxed);
            for byte in &mut buf[total..] {
                *byte = 0;
            }
            Ok(())
        })
        .map_err(|()| {
            take_stream_error(
                &first_error,
                "parallel stream reader error was not reported",
            )
        })?;

    Ok((
        !any_data.load(Ordering::Relaxed),
        actual_len.load(Ordering::Relaxed),
    ))
}

/// Parallel version of `read_block_all` — reads all shards concurrently.
fn read_block_all_par<R: std::io::Read + Send>(
    readers: &mut [R],
    buffers: &mut [Vec<u8>],
    max_len: usize,
) -> Result<(bool, usize), StreamError> {
    read_block_par(readers, buffers, max_len)
}

/// Parallel version of `write_block` — writes all shards concurrently.
fn write_block_par<W: std::io::Write + Send>(
    writers: &mut [W],
    buffers: &[Vec<u8>],
    len: usize,
    shard_offset: usize,
) -> Result<(), StreamError> {
    use rayon::prelude::*;

    let first_error: std::sync::Mutex<Option<StreamError>> = std::sync::Mutex::new(None);

    writers
        .par_iter_mut()
        .zip(buffers.par_iter())
        .enumerate()
        .try_for_each(|(i, (writer, buf))| {
            let mut written = 0;
            while written < len {
                match writer.write(&buf[written..len]) {
                    Ok(0) => {
                        if let Ok(mut guard) = first_error.lock()
                            && guard.is_none()
                        {
                            *guard = Some(StreamError::write(
                                shard_offset + i,
                                std::io::Error::new(
                                    std::io::ErrorKind::WriteZero,
                                    "write returned 0",
                                ),
                            ));
                        }
                        return Err(());
                    }
                    Ok(n) => written += n,
                    Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                    Err(e) => {
                        if let Ok(mut guard) = first_error.lock()
                            && guard.is_none()
                        {
                            *guard = Some(StreamError::write(shard_offset + i, e));
                        }
                        return Err(());
                    }
                }
            }
            Ok(())
        })
        .map_err(|()| {
            take_stream_error(
                &first_error,
                "parallel stream writer error was not reported",
            )
        })
}

// ---------------------------------------------------------------------------
// ReedSolomon streaming methods
// ---------------------------------------------------------------------------

impl super::ReedSolomon<crate::galois_8::Field> {
    /// Stream-encode data from readers into parity writers.
    ///
    /// Reads data shards in blocks of `options.block_size` bytes, encodes
    /// each block, and writes the resulting parity blocks.  This avoids
    /// loading the entire dataset into memory.
    ///
    /// # Errors
    ///
    /// Returns [`StreamError`] on I/O failure or codec error.
    pub fn encode_stream(
        &self,
        data: &mut [impl std::io::Read + Send],
        parity: &mut [impl std::io::Write + Send],
        options: &StreamOptions,
    ) -> Result<(), StreamError> {
        let block_size = options.block_size;
        let data_count = self.data_shard_count;
        let parity_count = self.parity_shard_count;

        debug_assert_eq!(data.len(), data_count);
        debug_assert_eq!(parity.len(), parity_count);

        let mut data_bufs: Vec<Vec<u8>> = (0..data_count)
            .map(|_| Vec::with_capacity(block_size))
            .collect();
        let mut parity_bufs: Vec<Vec<u8>> = (0..parity_count)
            .map(|_| Vec::with_capacity(block_size))
            .collect();

        loop {
            let (all_eof, actual_len) = read_block_par(data, &mut data_bufs, block_size)?;
            if all_eof {
                break;
            }

            // Resize parity buffers to match actual length.
            for buf in parity_bufs.iter_mut() {
                buf.resize(actual_len, 0);
            }

            // Encode (parallel codec).
            let data_refs: Vec<&[u8]> = data_bufs.iter().map(|b| &b[..actual_len]).collect();
            let mut parity_refs: Vec<&mut [u8]> = parity_bufs
                .iter_mut()
                .map(|b| &mut b[..actual_len])
                .collect();

            self.encode_sep_par(&data_refs, &mut parity_refs)
                .map_err(|e| StreamError::codec(0, e))?;

            // Write parity (parallel I/O).
            write_block_par(parity, &parity_bufs, actual_len, data_count)?;
        }

        Ok(())
    }

    /// Stream-verify data + parity from readers.
    ///
    /// Reads all shards in blocks, verifying each block independently.
    /// Returns `Ok(true)` if every block is valid, `Ok(false)` if any block
    /// fails verification.
    pub fn verify_stream(
        &self,
        shards: &mut [impl std::io::Read + Send],
        options: &StreamOptions,
    ) -> Result<bool, StreamError> {
        let block_size = options.block_size;
        let total = self.total_shard_count;

        debug_assert_eq!(shards.len(), total);

        let mut bufs: Vec<Vec<u8>> = (0..total).map(|_| Vec::with_capacity(block_size)).collect();

        loop {
            let (all_eof, actual_len) = read_block_all_par(shards, &mut bufs, block_size)?;
            if all_eof {
                break;
            }

            let refs: Vec<&[u8]> = bufs.iter().map(|b| &b[..actual_len]).collect();

            let valid = self
                .verify_par(&refs)
                .map_err(|e| StreamError::codec(0, e))?;
            if !valid {
                return Ok(false);
            }
        }

        Ok(true)
    }

    /// Stream-reconstruct missing shards.
    ///
    /// `shards` has one entry per total shard.  Present shards contain their
    /// data in a `Cursor<Vec<u8>>`; missing shards use an empty cursor
    /// (`Cursor::new(Vec::new())`).  Recovered data is written into the
    /// missing shards' cursors.
    ///
    /// The function reads blocks from present shards, reconstructs missing
    /// blocks, and writes recovered data into the missing cursors.  The set
    /// of missing shard indices must be consistent across all blocks.
    ///
    /// # Limitations
    ///
    /// For Leopard-family codecs the entire dataset must fit in memory; this
    /// streaming path is only supported for the classic Reed-Solomon family.
    pub fn reconstruct_stream(
        &self,
        shards: &mut [std::io::Cursor<Vec<u8>>],
        options: &StreamOptions,
    ) -> Result<(), StreamError> {
        let block_size = options.block_size;
        let total = self.total_shard_count;

        debug_assert_eq!(shards.len(), total);

        // Determine which shards are present (non-empty cursor).
        let mut present = vec![false; total];
        for (i, shard) in shards.iter().enumerate() {
            present[i] = !shard.get_ref().is_empty();
        }

        let missing_count = present.iter().filter(|&&p| !p).count();
        if missing_count > self.parity_shard_count {
            return Err(StreamError::codec(0, crate::Error::TooFewShardsPresent));
        }

        // Strategy: read present shards into buffers per block, reconstruct,
        // then write recovered data into missing shards' cursors.

        // Pre-allocate read buffers for present shards only.
        let mut bufs: Vec<Vec<u8>> = (0..total)
            .map(|i| {
                if present[i] {
                    Vec::with_capacity(block_size)
                } else {
                    Vec::new()
                }
            })
            .collect();

        loop {
            // Pass 1: read from present shards into buffers (parallel).
            use rayon::prelude::*;
            use std::sync::atomic::{AtomicBool, Ordering};

            let any_data = AtomicBool::new(false);
            let first_error: std::sync::Mutex<Option<StreamError>> = std::sync::Mutex::new(None);

            #[allow(clippy::type_complexity)]
            let mut indexed: Vec<(usize, &mut std::io::Cursor<Vec<u8>>, &mut Vec<u8>)> = shards
                .iter_mut()
                .zip(bufs.iter_mut())
                .enumerate()
                .filter(|(i, _)| present[*i])
                .map(|(i, (s, b))| (i, s, b))
                .collect();

            indexed
                .par_iter_mut()
                .try_for_each(|(i, shard, buf)| {
                    buf.resize(block_size, 0);
                    let mut total_read = 0;
                    while total_read < block_size {
                        match shard.read(&mut buf[total_read..]) {
                            Ok(0) => break,
                            Ok(n) => {
                                total_read += n;
                                any_data.store(true, Ordering::Relaxed);
                            }
                            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                            Err(e) => {
                                if let Ok(mut guard) = first_error.lock()
                                    && guard.is_none()
                                {
                                    *guard = Some(StreamError::read(*i, e));
                                }
                                return Err(());
                            }
                        }
                    }
                    buf.truncate(total_read);
                    Ok(())
                })
                .map_err(|()| {
                    take_stream_error(
                        &first_error,
                        "parallel stream reader error was not reported",
                    )
                })?;

            if !any_data.load(Ordering::Relaxed) {
                break;
            }

            // Compute actual_len from present shards only.
            let actual_len = bufs
                .iter()
                .enumerate()
                .filter(|(idx, _)| present[*idx])
                .find_map(|(_, b)| if b.is_empty() { None } else { Some(b.len()) })
                .unwrap_or(0);
            if actual_len == 0 {
                break;
            }

            // Zero-pad present shards to actual_len, fill missing shards.
            for buf in bufs.iter_mut() {
                buf.resize(actual_len, 0);
            }

            let mut reconstruct_bufs: Vec<Option<Vec<u8>>> = (0..total)
                .map(|i| {
                    if present[i] {
                        Some(mem::take(&mut bufs[i]))
                    } else {
                        None
                    }
                })
                .collect();

            self.reconstruct(&mut reconstruct_bufs)
                .map_err(|e| StreamError::codec(0, e))?;

            // Write recovered data into missing shards' cursors.
            // (reconstruct fills in all missing shards — data and parity)
            for (i, shard) in shards.iter_mut().enumerate() {
                if !present[i] {
                    let recovered = reconstruct_bufs[i].as_ref().expect("missing shard buffer");
                    shard.get_mut().extend_from_slice(&recovered[..actual_len]);
                }
            }
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::galois_8::ReedSolomon;

    fn make_codec(data: usize, parity: usize) -> ReedSolomon {
        ReedSolomon::new(data, parity).unwrap()
    }

    fn random_data(len: usize) -> Vec<u8> {
        // Simple deterministic fill for reproducibility.
        (0..len)
            .map(|i| (i.wrapping_mul(73).wrapping_add(17)) as u8)
            .collect()
    }

    #[test]
    fn test_encode_stream_basic() {
        let rs = make_codec(3, 2);
        let shard_size = 4096;

        let d0 = random_data(shard_size);
        let d1 = random_data(shard_size);
        let d2 = random_data(shard_size);

        let mut readers: Vec<&[u8]> = vec![d0.as_slice(), d1.as_slice(), d2.as_slice()];
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(), Vec::new()];

        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        assert_eq!(writers[0].len(), shard_size);
        assert_eq!(writers[1].len(), shard_size);

        // Verify parity is correct by verifying the full shard set.
        let all: Vec<&[u8]> = vec![
            d0.as_slice(),
            d1.as_slice(),
            d2.as_slice(),
            writers[0].as_slice(),
            writers[1].as_slice(),
        ];
        assert!(rs.verify(&all).unwrap());
    }

    #[test]
    fn test_encode_stream_multi_block() {
        let rs = make_codec(3, 2);
        let total_size = 10 * 1024; // 10 KiB
        let block_size = 4096;

        let d0 = random_data(total_size);
        let d1 = random_data(total_size);
        let d2 = random_data(total_size);

        let mut readers: Vec<&[u8]> = vec![d0.as_slice(), d1.as_slice(), d2.as_slice()];
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(), Vec::new()];

        let opts = StreamOptions::new().with_block_size(block_size);
        rs.encode_stream(&mut readers, &mut writers, &opts).unwrap();

        assert_eq!(writers[0].len(), total_size);
        assert_eq!(writers[1].len(), total_size);
    }

    #[test]
    fn test_encode_stream_empty() {
        let rs = make_codec(3, 2);
        let empty: Vec<&[u8]> = vec![&[], &[], &[]];
        let mut readers = empty.clone();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(), Vec::new()];

        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Empty input → empty output.
        assert!(writers[0].is_empty());
        assert!(writers[1].is_empty());
    }

    #[test]
    fn test_encode_stream_unequal_lengths() {
        let rs = make_codec(3, 2);

        let d0 = random_data(1000);
        let d1 = random_data(500);
        let d2 = random_data(800);

        let mut readers: Vec<&[u8]> = vec![d0.as_slice(), d1.as_slice(), d2.as_slice()];
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(), Vec::new()];

        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // All outputs should be 1000 bytes (max of input lengths, since we
        // zero-pad short shards).
        assert_eq!(writers[0].len(), 1000);
        assert_eq!(writers[1].len(), 1000);
    }

    #[test]
    fn test_encode_stream_10x4() {
        let rs = make_codec(10, 4);
        let shard_size = 64 * 1024;

        let data: Vec<Vec<u8>> = (0..10).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 4];

        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Verify.
        let mut all: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        for w in &writers {
            all.push(w.as_slice());
        }
        assert!(rs.verify(&all).unwrap());
    }

    #[test]
    fn test_verify_stream_valid() {
        let rs = make_codec(3, 2);
        let shard_size = 4096;

        let d = vec![random_data(shard_size); 3];
        let mut readers: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Build full shard set for verification.
        let mut all_data: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        for w in &writers {
            all_data.push(w.as_slice());
        }
        let mut all_readers: Vec<&[u8]> = all_data;

        assert!(
            rs.verify_stream(&mut all_readers, &StreamOptions::default())
                .unwrap()
        );
    }

    #[test]
    fn test_verify_stream_corrupted() {
        let rs = make_codec(3, 2);
        let shard_size = 4096;

        let d = vec![random_data(shard_size); 3];
        let mut readers: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Corrupt a parity shard.
        writers[0][0] ^= 0xFF;

        let mut all_data: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        for w in &writers {
            all_data.push(w.as_slice());
        }
        let mut all_readers: Vec<&[u8]> = all_data;

        assert!(
            !rs.verify_stream(&mut all_readers, &StreamOptions::default())
                .unwrap()
        );
    }

    #[test]
    fn test_reconstruct_stream_single_missing() {
        let rs = make_codec(3, 2);
        let shard_size = 4096;

        let d: Vec<Vec<u8>> = (0..3).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        let mut parity_writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut parity_writers, &StreamOptions::default())
            .unwrap();

        // data[0] is missing — use empty Cursor.
        let mut shards: Vec<std::io::Cursor<Vec<u8>>> = vec![
            std::io::Cursor::new(Vec::new()), // missing
            std::io::Cursor::new(d[1].clone()),
            std::io::Cursor::new(d[2].clone()),
            std::io::Cursor::new(parity_writers[0].clone()),
            std::io::Cursor::new(parity_writers[1].clone()),
        ];

        rs.reconstruct_stream(&mut shards, &StreamOptions::default())
            .unwrap();

        // Shard 0 should have been recovered into the cursor's inner Vec.
        assert_eq!(shards[0].get_ref(), &d[0]);
    }

    #[test]
    fn test_reconstruct_non_streaming() {
        // Verify basic encode + reconstruct works without streaming.
        let rs = make_codec(3, 2);
        let shard_size = 4096;

        let d: Vec<Vec<u8>> = (0..3).map(|_| random_data(shard_size)).collect();
        let data_refs: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        let mut p0 = vec![0u8; shard_size];
        let mut p1 = vec![0u8; shard_size];
        let mut parity_refs: Vec<&mut [u8]> = vec![&mut p0, &mut p1];
        rs.encode_sep(&data_refs, &mut parity_refs).unwrap();

        // Now reconstruct with shard 0 missing.
        let mut shards: Vec<Option<Vec<u8>>> = vec![
            None,
            Some(d[1].clone()),
            Some(d[2].clone()),
            Some(p0.clone()),
            Some(p1.clone()),
        ];
        rs.reconstruct(&mut shards).unwrap();

        assert_eq!(shards[0].as_ref().unwrap(), &d[0]);
    }

    #[test]
    fn test_reconstruct_stream_basic() {
        let rs = make_codec(3, 2);
        let shard_size = 64;

        let d: Vec<Vec<u8>> = (0..3).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = d.iter().map(|s| s.as_slice()).collect();
        let mut parity_writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut parity_writers, &StreamOptions::default())
            .unwrap();

        // Verify parity is correct.
        let all: Vec<&[u8]> = vec![
            d[0].as_slice(),
            d[1].as_slice(),
            d[2].as_slice(),
            parity_writers[0].as_slice(),
            parity_writers[1].as_slice(),
        ];
        assert!(rs.verify(&all).unwrap());

        // Missing shard 0 — empty Cursor; present shards have data.
        let mut shards: Vec<std::io::Cursor<Vec<u8>>> = vec![
            std::io::Cursor::new(Vec::new()), // missing
            std::io::Cursor::new(d[1].clone()),
            std::io::Cursor::new(d[2].clone()),
            std::io::Cursor::new(parity_writers[0].clone()),
            std::io::Cursor::new(parity_writers[1].clone()),
        ];

        rs.reconstruct_stream(&mut shards, &StreamOptions::default())
            .unwrap();

        // Shard 0 recovered into the empty cursor's inner Vec.
        let recovered = shards[0].get_ref();
        assert_eq!(
            recovered.len(),
            d[0].len(),
            "recovered len {} != expected len {}",
            recovered.len(),
            d[0].len()
        );
        assert_eq!(
            recovered,
            &d[0],
            "recovered: {:?}, expected: {:?}",
            &recovered[..8],
            &d[0][..8]
        );
    }

    #[test]
    fn test_stream_options_builder() {
        let opts = StreamOptions::new().with_block_size(1024 * 1024);
        assert_eq!(opts.block_size, 1024 * 1024);

        // Minimum is 1 KiB.
        let opts = StreamOptions::new().with_block_size(100);
        assert_eq!(opts.block_size, 1024);
    }

    #[test]
    fn test_stream_error_display() {
        let e = StreamError::read(
            3,
            std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "eof"),
        );
        let s = format!("{e}");
        assert!(s.contains("shard 3"));
        assert!(s.contains("eof"));

        let e = StreamError::codec(0, crate::Error::TooFewShardsPresent);
        let s = format!("{e}");
        assert!(s.contains("codec"));
    }

    // -----------------------------------------------------------------------
    // Concurrent stream tests (P0-2e-3)
    // -----------------------------------------------------------------------

    #[test]
    fn test_encode_stream_concurrent() {
        let rs = make_codec(4, 2);
        let shard_size = 8 * 1024;

        let data: Vec<Vec<u8>> = (0..4).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 2];

        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Verify parity correctness.
        let mut all: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        for w in &writers {
            all.push(w.as_slice());
        }
        assert!(rs.verify(&all).unwrap());
    }

    #[test]
    fn test_verify_stream_concurrent() {
        let rs = make_codec(4, 2);
        let shard_size = 8 * 1024;

        let data: Vec<Vec<u8>> = (0..4).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();

        // Valid case.
        let mut all: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        for w in &writers {
            all.push(w.as_slice());
        }
        let mut all_readers: Vec<&[u8]> = all;
        assert!(
            rs.verify_stream(&mut all_readers, &StreamOptions::default())
                .unwrap()
        );

        // Corrupted case.
        let mut corrupted = writers[0].clone();
        corrupted[0] ^= 0xFF;
        let mut all_corrupt: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        all_corrupt.push(corrupted.as_slice());
        all_corrupt.push(writers[1].as_slice());
        assert!(
            !rs.verify_stream(&mut all_corrupt, &StreamOptions::default())
                .unwrap()
        );
    }

    #[test]
    fn test_reconstruct_stream_concurrent() {
        let rs = make_codec(4, 2);
        let shard_size = 8 * 1024;

        let data: Vec<Vec<u8>> = (0..4).map(|_| random_data(shard_size)).collect();
        let mut readers: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut parity_writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        rs.encode_stream(&mut readers, &mut parity_writers, &StreamOptions::default())
            .unwrap();

        // Missing shards: data[1] and parity[0].
        let mut shards: Vec<std::io::Cursor<Vec<u8>>> = vec![
            std::io::Cursor::new(data[0].clone()),
            std::io::Cursor::new(Vec::new()), // missing
            std::io::Cursor::new(data[2].clone()),
            std::io::Cursor::new(data[3].clone()),
            std::io::Cursor::new(Vec::new()), // missing
            std::io::Cursor::new(parity_writers[1].clone()),
        ];

        rs.reconstruct_stream(&mut shards, &StreamOptions::default())
            .unwrap();

        assert_eq!(shards[1].get_ref(), &data[1]);
    }

    #[test]
    fn test_concurrent_stream_large_blocks() {
        let rs = make_codec(10, 4);
        let total_size = 1024 * 1024; // 1 MiB
        let block_size = 256 * 1024; // 256 KiB blocks

        let data: Vec<Vec<u8>> = (0..10).map(|_| random_data(total_size)).collect();
        let mut readers: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 4];

        let opts = StreamOptions::new().with_block_size(block_size);
        rs.encode_stream(&mut readers, &mut writers, &opts).unwrap();

        // Verify all blocks.
        let mut all: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        for w in &writers {
            all.push(w.as_slice());
        }
        assert!(rs.verify(&all).unwrap());

        // Reconstruct with 2 missing data shards.
        let mut shards: Vec<std::io::Cursor<Vec<u8>>> = Vec::new();
        for d in &data {
            shards.push(std::io::Cursor::new(d.clone()));
        }
        shards[0] = std::io::Cursor::new(Vec::new());
        shards[5] = std::io::Cursor::new(Vec::new());
        for w in &writers {
            shards.push(std::io::Cursor::new(w.clone()));
        }

        rs.reconstruct_stream(&mut shards, &StreamOptions::default())
            .unwrap();

        assert_eq!(shards[0].get_ref(), &data[0]);
        assert_eq!(shards[5].get_ref(), &data[5]);
    }

    #[test]
    fn test_encode_stream_zero_length() {
        let rs = ReedSolomon::new(3, 2).unwrap();
        let mut readers: Vec<&[u8]> = vec![b""; 3];
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 2];
        // Zero-length data should succeed (encode produces empty parity).
        rs.encode_stream(&mut readers, &mut writers, &StreamOptions::default())
            .unwrap();
        assert!(writers.iter().all(|w| w.is_empty()));
    }

    #[test]
    fn test_encode_stream_single_byte_block() {
        let rs = ReedSolomon::new(2, 1).unwrap();
        let d0 = vec![0xABu8; 4];
        let d1 = vec![0xCDu8; 4];
        let mut readers: Vec<&[u8]> = vec![d0.as_slice(), d1.as_slice()];
        let mut writers: Vec<Vec<u8>> = vec![Vec::new(); 1];
        let opts = StreamOptions::new().with_block_size(1);
        rs.encode_stream(&mut readers, &mut writers, &opts).unwrap();

        // Verify with the smallest possible block size.
        let mut all: Vec<&[u8]> = vec![d0.as_slice(), d1.as_slice()];
        for w in &writers {
            all.push(w.as_slice());
        }
        assert!(rs.verify(&all).unwrap());
    }

    #[test]
    fn test_reconstruct_stream_minimum_present() {
        let rs = ReedSolomon::new(3, 2).unwrap();
        let shard_len = 16usize;
        let data: Vec<Vec<u8>> = (0..3).map(|i| vec![i as u8 + 1; shard_len]).collect();

        // Encode to get parity.
        let refs: Vec<&[u8]> = data.iter().map(|d| d.as_slice()).collect();
        let mut parity = vec![vec![0u8; shard_len]; 2];
        let mut par_refs: Vec<&mut [u8]> = parity.iter_mut().map(|p| &mut p[..]).collect();
        rs.encode_sep(&refs, &mut par_refs).unwrap();

        // Keep minimum viable: 3 of 5 shards (data_shard_count).
        // Present: shard 0 (data), shard 3 (parity), shard 4 (parity).
        // Missing: shard 1 (data), shard 2 (data).
        let mut shards: Vec<std::io::Cursor<Vec<u8>>> = vec![
            std::io::Cursor::new(data[0].clone()),
            std::io::Cursor::new(Vec::new()),
            std::io::Cursor::new(Vec::new()),
            std::io::Cursor::new(parity[0].clone()),
            std::io::Cursor::new(parity[1].clone()),
        ];

        rs.reconstruct_stream(&mut shards, &StreamOptions::default())
            .unwrap();

        // Verify reconstructed data shards match originals.
        assert_eq!(shards[1].get_ref(), &data[1], "shard 1 mismatch");
        assert_eq!(shards[2].get_ref(), &data[2], "shard 2 mismatch");
    }
}