fix bitrot

ecstore/src/bitrot.rs

@@ -28,7 +28,7 @@ pub async fn create_bitrot_reader(
     checksum_algo: HashAlgorithm,
 ) -> disk::error::Result<Option<BitrotReader<Box<dyn AsyncRead + Send + Sync + Unpin>>>> {
     // Calculate the total length to read, including the checksum overhead
-    let length = offset.div_ceil(shard_size) * checksum_algo.size() + length;
+    let length = length.div_ceil(shard_size) * checksum_algo.size() + length;
 
     if let Some(data) = inline_data {
         // Use inline data

ecstore/src/disk/local.rs

@@ -1399,8 +1399,6 @@ impl DiskAPI for LocalDisk {
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: i64) -> Result<FileWriter> {
-        // warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path);
-
         if !origvolume.is_empty() {
             let origvolume_dir = self.get_bucket_path(origvolume)?;
             if !skip_access_checks(origvolume) {
@@ -1431,8 +1429,6 @@ impl DiskAPI for LocalDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     // async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<File> {
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
-        warn!("disk append_file:  volume: {}, path: {}", volume, path);
-
         let volume_dir = self.get_bucket_path(volume)?;
         if !skip_access_checks(volume) {
             access(&volume_dir)
@@ -1497,7 +1493,9 @@ impl DiskAPI for LocalDisk {
             return Err(DiskError::FileCorrupt);
         }
 
-        f.seek(SeekFrom::Start(offset as u64)).await?;
+        if offset > 0 {
+            f.seek(SeekFrom::Start(offset as u64)).await?;
+        }
 
         Ok(Box::new(f))
     }

ecstore/src/disk/remote.rs

@@ -608,7 +608,14 @@ impl DiskAPI for RemoteDisk {
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
-        info!("read_file_stream {}/{}/{}", self.endpoint.to_string(), volume, path);
+        // warn!(
+        //     "disk remote read_file_stream {}/{}/{} offset={} length={}",
+        //     self.endpoint.to_string(),
+        //     volume,
+        //     path,
+        //     offset,
+        //     length
+        // );
         let url = format!(
             "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
             self.endpoint.grid_host(),
@@ -641,7 +648,13 @@ impl DiskAPI for RemoteDisk {
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter> {
-        info!("create_file {}/{}/{}", self.endpoint.to_string(), volume, path);
+        // warn!(
+        //     "disk remote create_file {}/{}/{} file_size={}",
+        //     self.endpoint.to_string(),
+        //     volume,
+        //     path,
+        //     file_size
+        // );
 
         let url = format!(
             "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",

ecstore/src/erasure_coding/bitrot.rs

@@ -1,7 +1,9 @@
 use bytes::Bytes;
 use pin_project_lite::pin_project;
-use rustfs_utils::{HashAlgorithm, read_full, write_all};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
+use rustfs_utils::HashAlgorithm;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tracing::error;
+use uuid::Uuid;
 
 pin_project! {
     /// BitrotReader reads (hash+data) blocks from an async reader and verifies hash integrity.
@@ -12,10 +14,11 @@ pin_project! {
         shard_size: usize,
         buf: Vec<u8>,
         hash_buf: Vec<u8>,
-        hash_read: usize,
-        data_buf: Vec<u8>,
-        data_read: usize,
-        hash_checked: bool,
+        // hash_read: usize,
+        // data_buf: Vec<u8>,
+        // data_read: usize,
+        // hash_checked: bool,
+        id: Uuid,
     }
 }
 
@@ -32,10 +35,11 @@ where
             shard_size,
             buf: Vec::new(),
             hash_buf: vec![0u8; hash_size],
-            hash_read: 0,
-            data_buf: Vec::new(),
-            data_read: 0,
-            hash_checked: false,
+            // hash_read: 0,
+            // data_buf: Vec::new(),
+            // data_read: 0,
+            // hash_checked: false,
+            id: Uuid::new_v4(),
         }
     }
 
@@ -51,30 +55,31 @@ where
 
         let hash_size = self.hash_algo.size();
         // Read hash
-        let mut hash_buf = vec![0u8; hash_size];
+
         if hash_size > 0 {
-            self.inner.read_exact(&mut hash_buf).await?;
+            self.inner.read_exact(&mut self.hash_buf).await.map_err(|e| {
+                error!("bitrot reader read hash error: {}", e);
+                e
+            })?;
         }
 
-        let data_len = read_full(&mut self.inner, out).await?;
-
-        // // Read data
-        // let mut data_len = 0;
-        // while data_len < out.len() {
-        //     let n = self.inner.read(&mut out[data_len..]).await?;
-        //     if n == 0 {
-        //         break;
-        //     }
-        //     data_len += n;
-        //     // Only read up to one shard_size block
-        //     if data_len >= self.shard_size {
-        //         break;
-        //     }
-        // }
+        // Read data
+        let mut data_len = 0;
+        while data_len < out.len() {
+            let n = self.inner.read(&mut out[data_len..]).await.map_err(|e| {
+                error!("bitrot reader read data error: {}", e);
+                e
+            })?;
+            if n == 0 {
+                break;
+            }
+            data_len += n;
+        }
 
         if hash_size > 0 {
             let actual_hash = self.hash_algo.hash_encode(&out[..data_len]);
-            if actual_hash.as_ref() != hash_buf.as_slice() {
+            if actual_hash.as_ref() != self.hash_buf.as_slice() {
+                error!("bitrot reader hash mismatch, id={} data_len={}, out_len={}", self.id, data_len, out.len());
                 return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch"));
             }
         }
@@ -145,22 +150,20 @@ where
 
         self.buf.extend_from_slice(buf);
 
-        // Write hash+data in one call
-        let mut n = write_all(&mut self.inner, &self.buf).await?;
+        self.inner.write_all(&self.buf).await?;
 
-        if n < hash_algo.size() {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::WriteZero,
-                "short write: not enough bytes written",
-            ));
-        }
+        self.inner.flush().await?;
 
-        n -= hash_algo.size();
+        let n = self.buf.len();
 
         self.buf.clear();
 
         Ok(n)
     }
+
+    pub async fn shutdown(&mut self) -> std::io::Result<()> {
+        self.inner.shutdown().await
+    }
 }
 
 pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: HashAlgorithm) -> usize {
@@ -330,6 +333,10 @@ impl BitrotWriterWrapper {
         self.bitrot_writer.write(buf).await
     }
 
+    pub async fn shutdown(&mut self) -> std::io::Result<()> {
+        self.bitrot_writer.shutdown().await
+    }
+
     /// Extract the inline buffer data, consuming the wrapper
     pub fn into_inline_data(self) -> Option<Vec<u8>> {
         match self.writer_type {

ecstore/src/erasure_coding/decode.rs

@@ -67,36 +67,34 @@ where
         }
 
         // 使用并发读取所有分片
+        let mut read_futs = Vec::with_capacity(self.readers.len());
 
-        let read_futs: Vec<_> = self
-            .readers
-            .iter_mut()
-            .enumerate()
-            .map(|(i, opt_reader)| {
-                if let Some(reader) = opt_reader.as_mut() {
+        for (i, opt_reader) in self.readers.iter_mut().enumerate() {
+            let future = if let Some(reader) = opt_reader.as_mut() {
+                Box::pin(async move {
                     let mut buf = vec![0u8; shard_size];
-                    // 需要move i, buf
-                    Some(async move {
-                        match reader.read(&mut buf).await {
-                            Ok(n) => {
-                                buf.truncate(n);
-                                (i, Ok(buf))
-                            }
-                            Err(e) => (i, Err(Error::from(e))),
+                    match reader.read(&mut buf).await {
+                        Ok(n) => {
+                            buf.truncate(n);
+                            (i, Ok(buf))
                         }
-                    })
-                } else {
-                    None
-                }
-            })
-            .collect();
+                        Err(e) => (i, Err(Error::from(e))),
+                    }
+                }) as std::pin::Pin<Box<dyn std::future::Future<Output = (usize, Result<Vec<u8>, Error>)> + Send>>
+            } else {
+                // reader是None时返回FileNotFound错误
+                Box::pin(async move { (i, Err(Error::FileNotFound)) })
+                    as std::pin::Pin<Box<dyn std::future::Future<Output = (usize, Result<Vec<u8>, Error>)> + Send>>
+            };
+            read_futs.push(future);
+        }
 
-        // 过滤掉None，join_all
-        let mut results = join_all(read_futs.into_iter().flatten()).await;
+        let results = join_all(read_futs).await;
 
         let mut shards: Vec<Option<Vec<u8>>> = vec![None; self.readers.len()];
         let mut errs = vec![None; self.readers.len()];
-        for (i, shard) in results.drain(..) {
+
+        for (i, shard) in results.into_iter() {
             match shard {
                 Ok(data) => {
                     if !data.is_empty() {

ecstore/src/erasure_coding/encode.rs

@@ -97,6 +97,13 @@ impl<'a> MultiWriter<'a> {
                 .join(", ")
         )))
     }
+
+    pub async fn _shutdown(&mut self) -> std::io::Result<()> {
+        for writer in self.writers.iter_mut().flatten() {
+            writer.shutdown().await?;
+        }
+        Ok(())
+    }
 }
 
 impl Erasure {
@@ -147,7 +154,7 @@ impl Erasure {
         }
 
         let (reader, total) = task.await??;
-
+        // writers.shutdown().await?;
         Ok((reader, total))
     }
 }

ecstore/src/erasure_coding/erasure.rs

@@ -555,6 +555,13 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_shard_file_size_cases2() {
+        let erasure = Erasure::new(12, 4, 1024 * 1024);
+
+        assert_eq!(erasure.shard_file_size(1572864), 131074);
+    }
+
     #[test]
     fn test_shard_file_size_cases() {
         let erasure = Erasure::new(4, 2, 8);
@@ -577,6 +584,8 @@ mod tests {
         assert_eq!(erasure.shard_file_size(1248739), 312186); // 1248739/8=156092, last=3, 3 div_ceil 4=1, 156092*2+1=312185
 
         assert_eq!(erasure.shard_file_size(43), 12); // 43/8=5, last=3, 3 div_ceil 4=1, 5*2+1=11
+
+        assert_eq!(erasure.shard_file_size(1572864), 393216); // 43/8=5, last=3, 3 div_ceil 4=1, 5*2+1=11
     }
 
     #[test]
@@ -677,9 +686,14 @@ mod tests {
 
     #[test]
     fn test_shard_file_offset() {
-        let erasure = Erasure::new(4, 2, 8);
-        let offset = erasure.shard_file_offset(0, 16, 32);
+        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]

ecstore/src/rebalance.rs

@@ -243,7 +243,7 @@ impl ECStore {
                     return Err(err);
                 }
 
-                error!("rebalanceMeta: not found, rebalance not started");
+                warn!("rebalanceMeta: not found, rebalance not started");
             }
         }
 
@@ -501,7 +501,7 @@ impl ECStore {
             if let Some(meta) = rebalance_meta.as_mut() {
                 meta.cancel = Some(tx)
             } else {
-                error!("start_rebalance: rebalance_meta is None exit");
+                warn!("start_rebalance: rebalance_meta is None exit");
                 return;
             }
 

ecstore/src/set_disk.rs

@@ -1859,8 +1859,6 @@ impl SetDisks {
 
         let (last_part_index, _) = fi.to_part_offset(end_offset)?;
 
-        // let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
-
         let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
 
         let mut total_readed = 0;