update filereader/writer todo

common/protos/src/generated/flatbuffers_generated/models.rs

@@ -29,7 +29,7 @@ pub mod models {
         #[inline]
         unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
             Self {
-                _tab: flatbuffers::Table::new(buf, loc),
+                _tab: unsafe { flatbuffers::Table::new(buf, loc) },
             }
         }
     }

crates/filemeta/src/fileinfo.rs

@@ -312,7 +312,7 @@ impl FileInfo {
 
     /// Check if the object is remote (transitioned to another tier)
     pub fn is_remote(&self) -> bool {
-        !self.transition_tier.as_ref().map_or(true, |s| s.is_empty())
+        !self.transition_tier.as_ref().is_none_or(|s| s.is_empty())
     }
 
     /// Get the data directory for this object

crates/rio/src/lib.rs

@@ -21,6 +21,7 @@ pub use hash_reader::*;
 pub mod compress;
 
 pub mod reader;
+pub use reader::WarpReader;
 
 mod writer;
 use tokio::io::{AsyncRead, BufReader};

crates/rio/src/reader.rs

@@ -1 +1,27 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, ReadBuf};
 
+use crate::{EtagResolvable, HashReaderDetector, Reader};
+
+pub struct WarpReader<R> {
+    inner: R,
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> WarpReader<R> {
+    pub fn new(inner: R) -> Self {
+        Self { inner }
+    }
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> AsyncRead for WarpReader<R> {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+impl<R: AsyncRead + Unpin + Send + Sync> HashReaderDetector for WarpReader<R> {}
+
+impl<R: AsyncRead + Unpin + Send + Sync> EtagResolvable for WarpReader<R> {}
+
+impl<R: AsyncRead + Unpin + Send + Sync> Reader for WarpReader<R> {}

crates/rio/src/writer.rs

@@ -1,8 +1,6 @@
 use std::io::Cursor;
 use std::pin::Pin;
-use std::task::{Context, Poll};
 use tokio::io::AsyncWrite;
-use tokio::io::AsyncWriteExt;
 
 use crate::HttpWriter;
 
@@ -92,77 +90,3 @@ impl AsyncWrite for Writer {
         }
     }
 }
-
-/// WriterAll wraps a Writer and ensures each write writes the entire buffer (like write_all).
-pub struct WriterAll<W: AsyncWrite + Unpin> {
-    inner: W,
-}
-
-impl<W: AsyncWrite + Unpin> WriterAll<W> {
-    pub fn new(inner: W) -> Self {
-        Self { inner }
-    }
-
-    /// Write the entire buffer, like write_all.
-    pub async fn write_all(&mut self, mut buf: &[u8]) -> std::io::Result<()> {
-        while !buf.is_empty() {
-            let n = self.inner.write(buf).await?;
-            if n == 0 {
-                return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "failed to write whole buffer"));
-            }
-            buf = &buf[n..];
-        }
-        Ok(())
-    }
-
-    /// Get a mutable reference to the inner writer.
-    pub fn get_mut(&mut self) -> &mut W {
-        &mut self.inner
-    }
-}
-
-impl<W: AsyncWrite + Unpin> AsyncWrite for WriterAll<W> {
-    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, mut buf: &[u8]) -> Poll<std::io::Result<usize>> {
-        let mut total_written = 0;
-        while !buf.is_empty() {
-            // Safety: W: Unpin
-            let inner_pin = Pin::new(&mut self.inner);
-            match inner_pin.poll_write(cx, buf) {
-                Poll::Ready(Ok(0)) => {
-                    if total_written == 0 {
-                        return Poll::Ready(Ok(0));
-                    } else {
-                        return Poll::Ready(Ok(total_written));
-                    }
-                }
-                Poll::Ready(Ok(n)) => {
-                    total_written += n;
-                    buf = &buf[n..];
-                }
-                Poll::Ready(Err(e)) => {
-                    if total_written == 0 {
-                        return Poll::Ready(Err(e));
-                    } else {
-                        return Poll::Ready(Ok(total_written));
-                    }
-                }
-                Poll::Pending => {
-                    if total_written == 0 {
-                        return Poll::Pending;
-                    } else {
-                        return Poll::Ready(Ok(total_written));
-                    }
-                }
-            }
-        }
-        Poll::Ready(Ok(total_written))
-    }
-
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        Pin::new(&mut self.inner).poll_flush(cx)
-    }
-
-    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        Pin::new(&mut self.inner).poll_shutdown(cx)
-    }
-}

ecstore/src/disk/error.rs

@@ -253,11 +253,11 @@ impl From<protos::proto_gen::node_service::Error> for DiskError {
     }
 }
 
-impl Into<protos::proto_gen::node_service::Error> for DiskError {
+impl From<DiskError> for protos::proto_gen::node_service::Error {
-    fn into(self) -> protos::proto_gen::node_service::Error {
+    fn from(e: DiskError) -> Self {
         protos::proto_gen::node_service::Error {
-            code: self.to_u32(),
+            code: e.to_u32(),
-            error_info: self.to_string(),
+            error_info: e.to_string(),
         }
     }
 }

ecstore/src/disk/error_reduce.rs

@@ -29,11 +29,7 @@ pub fn reduce_read_quorum_errs(errors: &[Option<Error>], ignored_errs: &[Error],
 
 pub fn reduce_quorum_errs(errors: &[Option<Error>], ignored_errs: &[Error], quorun: usize, quorun_err: Error) -> Option<Error> {
     let (max_count, err) = reduce_errs(errors, ignored_errs);
-    if max_count >= quorun {
+    if max_count >= quorun { err } else { Some(quorun_err) }
-        err
-    } else {
-        Some(quorun_err)
-    }
 }
 
 pub fn reduce_errs(errors: &[Option<Error>], ignored_errs: &[Error]) -> (usize, Option<Error>) {
@@ -59,7 +55,7 @@ pub fn reduce_errs(errors: &[Option<Error>], ignored_errs: &[Error]) -> (usize,
                 match (e1.to_string().as_str(), e2.to_string().as_str()) {
                     ("nil", _) => std::cmp::Ordering::Greater,
                     (_, "nil") => std::cmp::Ordering::Less,
-                    (a, b) => a.cmp(&b),
+                    (a, b) => a.cmp(b),
                 }
             } else {
                 count_cmp

ecstore/src/disk/local.rs

@@ -10,7 +10,6 @@ use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
 use crate::bucket::metadata_sys::{self};
 use crate::bucket::versioning::VersioningApi;
 use crate::bucket::versioning_sys::BucketVersioningSys;
-use crate::disk::STORAGE_FORMAT_FILE;
 use crate::disk::error::FileAccessDeniedWithContext;
 use crate::disk::error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error};
 use crate::disk::fs::{
@@ -19,8 +18,9 @@ use crate::disk::fs::{
 use crate::disk::os::{check_path_length, is_empty_dir};
 use crate::disk::{
     CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, CHECK_PART_VOLUME_NOT_FOUND,
-    conv_part_err_to_int,
+    FileReader, conv_part_err_to_int,
 };
+use crate::disk::{FileWriter, STORAGE_FORMAT_FILE};
 use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
 use crate::heal::data_scanner::{
     ScannerItem, ShouldSleepFn, SizeSummary, lc_has_active_rules, rep_has_active_rules, scan_data_folder,
@@ -30,7 +30,6 @@ use crate::heal::data_usage_cache::{DataUsageCache, DataUsageEntry};
 use crate::heal::error::{ERR_IGNORE_FILE_CONTRIB, ERR_SKIP_FILE};
 use crate::heal::heal_commands::{HealScanMode, HealingTracker};
 use crate::heal::heal_ops::HEALING_TRACKER_FILENAME;
-use crate::io::FileWriter;
 use crate::new_object_layer_fn;
 use crate::store_api::{ObjectInfo, StorageAPI};
 use crate::utils::os::get_info;
@@ -45,7 +44,7 @@ use rustfs_filemeta::{
     Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, Opts, RawFileInfo, UpdateFn, get_file_info,
     read_xl_meta_no_data,
 };
-use rustfs_rio::{Reader, bitrot_verify};
+use rustfs_rio::bitrot_verify;
 use rustfs_utils::HashAlgorithm;
 use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
@@ -1299,7 +1298,7 @@ impl DiskAPI for LocalDisk {
                 }
             }
 
-            remove_std(&dst_file_path).map_err(|e| to_file_error(e))?;
+            remove_std(&dst_file_path).map_err(to_file_error)?;
         }
 
         rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?;
@@ -1356,11 +1355,11 @@ impl DiskAPI for LocalDisk {
 
             if let Some(meta) = meta_op {
                 if !meta.is_dir() {
-                    return Err(DiskError::FileAccessDenied.into());
+                    return Err(DiskError::FileAccessDenied);
                 }
             }
 
-            remove(&dst_file_path).await.map_err(|e| to_file_error(e))?;
+            remove(&dst_file_path).await.map_err(to_file_error)?;
         }
 
         rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?;
@@ -1425,7 +1424,7 @@ impl DiskAPI for LocalDisk {
 
     // TODO: io verifier
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn read_file(&self, volume: &str, path: &str) -> Result<Box<dyn Reader>> {
+    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
         // warn!("disk read_file: volume: {}, path: {}", volume, path);
         let volume_dir = self.get_bucket_path(volume)?;
         if !skip_access_checks(volume) {
@@ -1443,7 +1442,7 @@ impl DiskAPI for LocalDisk {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Box<dyn Reader>> {
+    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
         // warn!(
         //     "disk read_file_stream: volume: {}, path: {}, offset: {}, length: {}",
         //     volume, path, offset, length
@@ -1615,7 +1614,7 @@ impl DiskAPI for LocalDisk {
                 let e: DiskError = to_file_error(e).into();
 
                 if e != DiskError::FileNotFound {
-                    return Err(e.into());
+                    return Err(e);
                 }
 
                 None

ecstore/src/disk/mod.rs

@@ -17,13 +17,10 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json";
 pub const STORAGE_FORMAT_FILE: &str = "xl.meta";
 pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
 
-use crate::{
+use crate::heal::{
-    heal::{
+    data_scanner::ShouldSleepFn,
-        data_scanner::ShouldSleepFn,
+    data_usage_cache::{DataUsageCache, DataUsageEntry},
-        data_usage_cache::{DataUsageCache, DataUsageEntry},
+    heal_commands::{HealScanMode, HealingTracker},
-        heal_commands::{HealScanMode, HealingTracker},
-    },
-    io::FileWriter,
 };
 use endpoint::Endpoint;
 use error::DiskError;
@@ -32,16 +29,21 @@ use local::LocalDisk;
 use madmin::info_commands::DiskMetrics;
 use remote::RemoteDisk;
 use rustfs_filemeta::{FileInfo, RawFileInfo};
-use rustfs_rio::Reader;
 use serde::{Deserialize, Serialize};
 use std::{fmt::Debug, path::PathBuf, sync::Arc};
 use time::OffsetDateTime;
-use tokio::{io::AsyncWrite, sync::mpsc::Sender};
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    sync::mpsc::Sender,
+};
 use tracing::warn;
 use uuid::Uuid;
 
 pub type DiskStore = Arc<Disk>;
 
+pub type FileReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
+pub type FileWriter = Box<dyn AsyncWrite + Send + Sync + Unpin>;
+
 #[derive(Debug)]
 pub enum Disk {
     Local(Box<LocalDisk>),
@@ -277,7 +279,7 @@ impl DiskAPI for Disk {
     }
 
     #[tracing::instrument(skip(self))]
-    async fn read_file(&self, volume: &str, path: &str) -> Result<Box<dyn Reader>> {
+    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
         match self {
             Disk::Local(local_disk) => local_disk.read_file(volume, path).await,
             Disk::Remote(remote_disk) => remote_disk.read_file(volume, path).await,
@@ -285,7 +287,7 @@ impl DiskAPI for Disk {
     }
 
     #[tracing::instrument(skip(self))]
-    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Box<dyn Reader>> {
+    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
         match self {
             Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await,
             Disk::Remote(remote_disk) => remote_disk.read_file_stream(volume, path, offset, length).await,
@@ -485,8 +487,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     // File operations.
     // 读目录下的所有文件、目录
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>>;
-    async fn read_file(&self, volume: &str, path: &str) -> Result<Box<dyn Reader>>;
+    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader>;
-    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Box<dyn Reader>>;
+    async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader>;
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
     async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter>;
     // ReadFileStream

ecstore/src/disk/os.rs

@@ -108,7 +108,7 @@ pub async fn rename_all(
 ) -> Result<()> {
     reliable_rename(src_file_path, dst_file_path.as_ref(), base_dir)
         .await
-        .map_err(|e| to_file_error(e))?;
+        .map_err(to_file_error)?;
 
     Ok(())
 }

ecstore/src/disk/remote.rs

@@ -13,30 +13,33 @@ use protos::{
 };
 use rmp_serde::Serializer;
 use rustfs_filemeta::{FileInfo, MetaCacheEntry, MetacacheWriter, RawFileInfo};
-use rustfs_rio::{HttpReader, Reader};
+use rustfs_rio::{HttpReader, HttpWriter};
 use serde::Serialize;
 use tokio::{
     io::AsyncWrite,
     sync::mpsc::{self, Sender},
 };
-use tokio_stream::{wrappers::ReceiverStream, StreamExt};
+use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 use tonic::Request;
 use tracing::info;
 use uuid::Uuid;
 
 use super::error::{Error, Result};
 use super::{
-    endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
+    CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions,
-    FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo,
+    ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
-    WalkDirOptions,
+    endpoint::Endpoint,
 };
 
-use crate::heal::{
+use crate::{
-    data_scanner::ShouldSleepFn,
+    disk::{FileReader, FileWriter},
-    data_usage_cache::{DataUsageCache, DataUsageEntry},
+    heal::{
-    heal_commands::{HealScanMode, HealingTracker},
+        data_scanner::ShouldSleepFn,
+        data_usage_cache::{DataUsageCache, DataUsageEntry},
+        heal_commands::{HealScanMode, HealingTracker},
+    },
 };
-use crate::io::{FileWriter, HttpFileWriter};
+
 use protos::proto_gen::node_service::RenamePartRequst;
 
 #[derive(Debug)]
@@ -552,7 +555,7 @@ impl DiskAPI for RemoteDisk {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn read_file(&self, volume: &str, path: &str) -> Result<Box<dyn Reader>> {
+    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
         info!("read_file {}/{}", volume, path);
 
         let url = format!(
@@ -569,7 +572,7 @@ 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<Box<dyn Reader>> {
+    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);
         let url = format!(
             "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
@@ -587,27 +590,35 @@ impl DiskAPI for RemoteDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
         info!("append_file {}/{}", volume, path);
-        Ok(Box::new(HttpFileWriter::new(
+
-            self.endpoint.grid_host().as_str(),
+        let url = format!(
-            self.endpoint.to_string().as_str(),
+            "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
-            volume,
+            self.endpoint.grid_host(),
-            path,
+            urlencoding::encode(self.endpoint.to_string().as_str()),
-            0,
+            urlencoding::encode(volume),
+            urlencoding::encode(path),
             true,
-        )?))
+            0
+        );
+
+        Ok(Box::new(HttpWriter::new(url, Method::PUT, HeaderMap::new()).await?))
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter> {
         info!("create_file {}/{}/{}", self.endpoint.to_string(), volume, path);
-        Ok(Box::new(HttpFileWriter::new(
+
-            self.endpoint.grid_host().as_str(),
+        let url = format!(
-            self.endpoint.to_string().as_str(),
+            "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}",
-            volume,
+            self.endpoint.grid_host(),
-            path,
+            urlencoding::encode(self.endpoint.to_string().as_str()),
-            file_size,
+            urlencoding::encode(volume),
+            urlencoding::encode(path),
             false,
-        )?))
+            file_size
+        );
+
+        Ok(Box::new(HttpWriter::new(url, Method::PUT, HeaderMap::new()).await?))
     }
 
     #[tracing::instrument(level = "debug", skip(self))]

ecstore/src/io.rs

@@ -6,9 +6,9 @@ use md5::Md5;
 use pin_project_lite::pin_project;
 use std::io;
 use std::pin::Pin;
-use std::task::ready;
 use std::task::Context;
 use std::task::Poll;
+use std::task::ready;
 use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tokio::io::ReadBuf;

ecstore/src/lib.rs

@@ -13,7 +13,7 @@ pub mod error;
 // pub mod file_meta_inline;
 pub mod global;
 pub mod heal;
-pub mod io;
+// pub mod io;
 // pub mod metacache;
 pub mod metrics_realtime;
 pub mod notification_sys;

ecstore/src/peer.rs

@@ -1,9 +1,9 @@
 use crate::disk::error::{Error, Result};
-use crate::disk::error_reduce::{is_all_buckets_not_found, reduce_write_quorum_errs, BUCKET_OP_IGNORED_ERRS};
+use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
 use crate::disk::{DiskAPI, DiskStore};
 use crate::global::GLOBAL_LOCAL_DISK_MAP;
 use crate::heal::heal_commands::{
-    HealOpts, DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET,
+    DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, HealOpts,
 };
 use crate::heal::heal_ops::RUSTFS_RESERVED_BUCKET;
 use crate::store::all_local_disk;
@@ -137,7 +137,7 @@ impl S3PeerSys {
                 return Ok(heal_bucket_results.read().await[i].clone());
             }
         }
-        Err(Error::VolumeNotFound.into())
+        Err(Error::VolumeNotFound)
     }
 
     pub async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
@@ -771,7 +771,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResu
             let bucket = bucket.to_string();
             let bs_clone = before_state.clone();
             let as_clone = after_state.clone();
-            let errs_clone = errs.iter().map(|e| e.as_ref().map(|e| e.clone())).collect::<Vec<_>>();
+            let errs_clone = errs.iter().map(|e| e.clone()).collect::<Vec<_>>();
             futures.push(async move {
                 if bs_clone.read().await[idx] == DRIVE_STATE_MISSING {
                     info!("bucket not find, will recreate");
@@ -785,7 +785,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResu
                         }
                     }
                 }
-                errs_clone[idx].as_ref().map(|e| e.clone())
+                errs_clone[idx].clone()
             });
         }
 

ecstore/src/pools.rs

@@ -1,13 +1,13 @@
 use crate::bucket::versioning_sys::BucketVersioningSys;
-use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions};
+use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
-use crate::config::com::{read_config, save_config, CONFIG_PREFIX};
+use crate::config::com::{CONFIG_PREFIX, read_config, save_config};
 use crate::disk::error::DiskError;
 use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
-use crate::error::{
-    is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_object_not_found,
-    is_err_version_not_found, StorageError,
-};
 use crate::error::{Error, Result};
+use crate::error::{
+    StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_object_not_found,
+    is_err_version_not_found,
+};
 use crate::heal::data_usage::DATA_USAGE_CACHE_NAME;
 use crate::heal::heal_commands::HealOpts;
 use crate::new_object_layer_fn;
@@ -16,7 +16,7 @@ use crate::set_disk::SetDisks;
 use crate::store_api::{
     BucketOptions, CompletePart, GetObjectReader, MakeBucketOptions, ObjectIO, ObjectOptions, PutObjReader, StorageAPI,
 };
-use crate::utils::path::{encode_dir_object, path_join, SLASH_SEPARATOR};
+use crate::utils::path::{SLASH_SEPARATOR, encode_dir_object, path_join};
 use crate::{sets::Sets, store::ECStore};
 use ::workers::workers::Workers;
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
@@ -25,7 +25,7 @@ use futures::future::BoxFuture;
 use http::HeaderMap;
 use rmp_serde::{Deserializer, Serializer};
 use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
-use rustfs_rio::{HashReader, Reader};
+use rustfs_rio::HashReader;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt::Display;
@@ -306,7 +306,7 @@ impl PoolMeta {
     }
 
     pub fn track_current_bucket_object(&mut self, idx: usize, bucket: String, object: String) {
-        if !self.pools.get(idx).is_some_and(|v| v.decommission.is_some()) {
+        if self.pools.get(idx).is_none_or(|v| v.decommission.is_none()) {
             return;
         }
 
@@ -319,7 +319,7 @@ impl PoolMeta {
     }
 
     pub async fn update_after(&mut self, idx: usize, pools: Vec<Arc<Sets>>, duration: Duration) -> Result<bool> {
-        if !self.pools.get(idx).is_some_and(|v| v.decommission.is_some()) {
+        if self.pools.get(idx).is_none_or(|v| v.decommission.is_none()) {
             return Err(Error::other("InvalidArgument"));
         }
 
@@ -882,7 +882,7 @@ impl ECStore {
         pool: Arc<Sets>,
         bi: DecomBucketInfo,
     ) -> Result<()> {
-        let wk = Workers::new(pool.disk_set.len() * 2).map_err(|v| Error::other(v))?;
+        let wk = Workers::new(pool.disk_set.len() * 2).map_err(Error::other)?;
 
         // let mut vc = None;
         // replication

ecstore/src/rebalance.rs

@@ -2,18 +2,18 @@ use std::io::Cursor;
 use std::sync::Arc;
 use std::time::SystemTime;
 
-use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions};
+use crate::StorageAPI;
+use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
 use crate::config::com::{read_config_with_metadata, save_config_with_opts};
 use crate::disk::error::DiskError;
-use crate::error::{is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found};
 use crate::error::{Error, Result};
+use crate::error::{is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found};
 use crate::global::get_global_endpoints;
 use crate::pools::ListCallback;
 use crate::set_disk::SetDisks;
 use crate::store::ECStore;
 use crate::store_api::{CompletePart, GetObjectReader, ObjectIO, ObjectOptions, PutObjReader};
 use crate::utils::path::encode_dir_object;
-use crate::StorageAPI;
 use common::defer;
 use http::HeaderMap;
 use rustfs_filemeta::{FileInfo, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
@@ -500,7 +500,7 @@ impl ECStore {
             if get_global_endpoints()
                 .as_ref()
                 .get(idx)
-                .map_or(true, |v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local))
+                .is_none_or(|v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local))
             {
                 warn!("start_rebalance: pool {} is not local, skipping", idx);
                 continue;
@@ -957,7 +957,7 @@ impl ECStore {
 
         let pool = self.pools[pool_index].clone();
 
-        let wk = Workers::new(pool.disk_set.len() * 2).map_err(|v| Error::other(v))?;
+        let wk = Workers::new(pool.disk_set.len() * 2).map_err(Error::other)?;
 
         for (set_idx, set) in pool.disk_set.iter().enumerate() {
             wk.clone().take().await;

ecstore/src/set_disk.rs

@@ -30,7 +30,6 @@ use crate::{
         },
         heal_ops::BG_HEALING_UUID,
     },
-    io::READ_BUFFER_SIZE,
     store_api::{
         BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
         ListMultipartsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartInfo, MultipartUploadResult, ObjectIO, ObjectInfo,
@@ -77,14 +76,13 @@ use std::time::SystemTime;
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
-    mem::replace,
     path::Path,
     sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
 use tokio::{
-    io::{AsyncWrite, empty},
+    io::AsyncWrite,
     sync::{RwLock, broadcast},
 };
 use tokio::{
@@ -97,6 +95,8 @@ use tracing::{debug, info, warn};
 use uuid::Uuid;
 use workers::workers::Workers;
 
+pub const DEFAULT_READ_BUFFER_SIZE: usize = 1024 * 1024;
+
 #[derive(Debug)]
 pub struct SetDisks {
     pub lockers: Vec<LockApi>,
@@ -146,12 +146,10 @@ impl SetDisks {
 
         disks.shuffle(&mut rng);
 
-        let disks = disks
+        disks
             .into_iter()
             .filter(|v| v.as_ref().is_some_and(|d| d.is_local()))
-            .collect();
+            .collect()
-
-        disks
     }
 
     pub async fn get_online_disks_with_healing(&self, incl_healing: bool) -> (Vec<DiskStore>, bool) {
@@ -248,12 +246,10 @@ impl SetDisks {
 
         disks.shuffle(&mut rng);
 
-        let disks = disks
+        disks
             .into_iter()
             .filter(|v| v.as_ref().is_some_and(|d| d.is_local()))
-            .collect();
+            .collect()
-
-        disks
     }
     fn default_write_quorum(&self) -> usize {
         let mut data_count = self.set_drive_count - self.default_parity_count;
@@ -446,7 +442,7 @@ impl SetDisks {
         let errs: Vec<Option<DiskError>> = join_all(futures)
             .await
             .into_iter()
-            .map(|e| e.unwrap_or_else(|_| Some(DiskError::Unexpected)))
+            .map(|e| e.unwrap_or(Some(DiskError::Unexpected)))
             .collect();
 
         if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
@@ -1890,7 +1886,11 @@ impl SetDisks {
                             till_offset,
                         )
                         .await?;
-                    let reader = BitrotReader::new(rd, erasure.shard_size(), HashAlgorithm::HighwayHash256);
+                    let reader = BitrotReader::new(
+                        Box::new(rustfs_rio::WarpReader::new(rd)),
+                        erasure.shard_size(),
+                        HashAlgorithm::HighwayHash256,
+                    );
                     readers.push(Some(reader));
                     errors.push(None);
                 } else {
@@ -1928,9 +1928,10 @@ impl SetDisks {
             //     "read part {} part_offset {},part_length {},part_size {}  ",
             //     part_number, part_offset, part_length, part_size
             // );
-            let (written, mut err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await;
+            let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await;
             if let Some(e) = err {
                 let de_err: DiskError = e.into();
+                let mut has_err = true;
                 if written == part_length {
                     match de_err {
                         DiskError::FileNotFound | DiskError::FileCorrupt => {
@@ -1947,14 +1948,16 @@ impl SetDisks {
                                     ..Default::default()
                                 })
                                 .await;
-                            err = None;
+                            has_err = false;
                         }
                         _ => {}
                     }
                 }
 
-                error!("erasure.decode err {} {:?}", written, &de_err);
+                if has_err {
-                return Err(de_err.into());
+                    error!("erasure.decode err {} {:?}", written, &de_err);
+                    return Err(de_err.into());
+                }
             }
 
             // debug!("ec decode {} writed size {}", part_number, n);
@@ -2500,27 +2503,35 @@ impl SetDisks {
 
                                         if let Some(ref data) = metadata.data {
                                             let rd = Cursor::new(data.clone());
-                                            let reader = BitrotReader::new(
+                                            let reader =
-                                                Box::new(rd),
+                                                BitrotReader::new(Box::new(rd), erasure.shard_size(), checksum_algo.clone());
-                                                erasure.shard_size(),
-                                                HashAlgorithm::HighwayHash256,
-                                            );
                                             readers.push(Some(reader));
                                             // errors.push(None);
                                         } else {
+                                            let length =
+                                                till_offset.div_ceil(erasure.shard_size()) * checksum_algo.size() + till_offset;
                                             let rd = match disk
-                                                .read_file(bucket, &format!("{}/{}/part.{}", object, src_data_dir, part.number))
+                                                .read_file_stream(
+                                                    bucket,
+                                                    &format!("{}/{}/part.{}", object, src_data_dir, part.number),
+                                                    0,
+                                                    length,
+                                                )
                                                 .await
                                             {
                                                 Ok(rd) => rd,
                                                 Err(e) => {
                                                     // errors.push(Some(e.into()));
+                                                    error!("heal_object read_file err: {:?}", e);
                                                     writers.push(None);
                                                     continue;
                                                 }
                                             };
-                                            let reader =
+                                            let reader = BitrotReader::new(
-                                                BitrotReader::new(rd, erasure.shard_size(), HashAlgorithm::HighwayHash256);
+                                                Box::new(rustfs_rio::WarpReader::new(rd)),
+                                                erasure.shard_size(),
+                                                HashAlgorithm::HighwayHash256,
+                                            );
                                             readers.push(Some(reader));
                                             // errors.push(None);
                                         }
@@ -3694,7 +3705,7 @@ impl SetDisks {
 
         let errs = join_all(futures).await.into_iter().map(|v| v.err()).collect::<Vec<_>>();
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS.as_ref(), write_quorum) {
+        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
             return Err(err);
         }
 
@@ -3749,7 +3760,7 @@ impl ObjectIO for SetDisks {
 
         // TODO: remote
 
-        let (rd, wd) = tokio::io::duplex(READ_BUFFER_SIZE);
+        let (rd, wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
 
         let (reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h)?;
 
@@ -5950,7 +5961,7 @@ mod tests {
         metadata.insert("etag".to_string(), "test-etag".to_string());
 
         let file_info = FileInfo {
-            metadata: metadata,
+            metadata,
             ..Default::default()
         };
         let parts_metadata = vec![file_info];

ecstore/src/store.rs

@@ -49,7 +49,6 @@ use glob::Pattern;
 use http::HeaderMap;
 use lazy_static::lazy_static;
 use madmin::heal_commands::HealResultItem;
-use rand::Rng;
 use rustfs_filemeta::MetaCacheEntry;
 use s3s::dto::{BucketVersioningStatus, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration};
 use std::cmp::Ordering;

rustfs/src/admin/rpc.rs

@@ -3,7 +3,7 @@ use super::router::Operation;
 use super::router::S3Router;
 use crate::storage::ecfs::bytes_stream;
 use ecstore::disk::DiskAPI;
-use ecstore::io::READ_BUFFER_SIZE;
+use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use ecstore::store::find_local_disk;
 use futures::TryStreamExt;
 use http::StatusCode;
@@ -72,7 +72,7 @@ impl Operation for ReadFile {
         Ok(S3Response::new((
             StatusCode::OK,
             Body::from(StreamingBlob::wrap(bytes_stream(
-                ReaderStream::with_capacity(file, READ_BUFFER_SIZE),
+                ReaderStream::with_capacity(file, DEFAULT_READ_BUFFER_SIZE),
                 query.length,
             ))),
         )))

rustfs/src/storage/ecfs.rs

@@ -30,8 +30,8 @@ use ecstore::bucket::tagging::decode_tags;
 use ecstore::bucket::tagging::encode_tags;
 use ecstore::bucket::versioning_sys::BucketVersioningSys;
 use ecstore::error::StorageError;
-use ecstore::io::READ_BUFFER_SIZE;
 use ecstore::new_object_layer_fn;
+use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use ecstore::store_api::BucketOptions;
 use ecstore::store_api::CompletePart;
 use ecstore::store_api::DeleteBucketOptions;
@@ -575,7 +575,7 @@ impl S3 for FS {
         let last_modified = info.mod_time.map(Timestamp::from);
 
         let body = Some(StreamingBlob::wrap(bytes_stream(
-            ReaderStream::with_capacity(reader.stream, READ_BUFFER_SIZE),
+            ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE),
             info.size,
         )));
 

s3select/api/src/object_store.rs

@@ -2,17 +2,16 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use chrono::Utc;
 use common::DEFAULT_DELIMITER;
-use ecstore::io::READ_BUFFER_SIZE;
+use ecstore::StorageAPI;
 use ecstore::new_object_layer_fn;
+use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use ecstore::store::ECStore;
 use ecstore::store_api::ObjectIO;
 use ecstore::store_api::ObjectOptions;
-use ecstore::StorageAPI;
 use futures::pin_mut;
 use futures::{Stream, StreamExt};
 use futures_core::stream::BoxStream;
 use http::HeaderMap;
-use object_store::path::Path;
 use object_store::Attributes;
 use object_store::GetOptions;
 use object_store::GetResult;
@@ -24,16 +23,17 @@ use object_store::PutMultipartOpts;
 use object_store::PutOptions;
 use object_store::PutPayload;
 use object_store::PutResult;
+use object_store::path::Path;
 use object_store::{Error as o_Error, Result};
 use pin_project_lite::pin_project;
+use s3s::S3Result;
 use s3s::dto::SelectObjectContentInput;
 use s3s::s3_error;
-use s3s::S3Result;
 use std::ops::Range;
 use std::pin::Pin;
 use std::sync::Arc;
-use std::task::ready;
 use std::task::Poll;
+use std::task::ready;
 use tokio::io::AsyncRead;
 use tokio_util::io::ReaderStream;
 use tracing::info;
@@ -117,14 +117,21 @@ impl ObjectStore for EcObjectStore {
         let payload = if self.need_convert {
             object_store::GetResultPayload::Stream(
                 bytes_stream(
-                    ReaderStream::with_capacity(ConvertStream::new(reader.stream, self.delimiter.clone()), READ_BUFFER_SIZE),
+                    ReaderStream::with_capacity(
+                        ConvertStream::new(reader.stream, self.delimiter.clone()),
+                        DEFAULT_READ_BUFFER_SIZE,
+                    ),
                     reader.object_info.size,
                 )
                 .boxed(),
             )
         } else {
             object_store::GetResultPayload::Stream(
-                bytes_stream(ReaderStream::with_capacity(reader.stream, READ_BUFFER_SIZE), reader.object_info.size).boxed(),
+                bytes_stream(
+                    ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE),
+                    reader.object_info.size,
+                )
+                .boxed(),
             )
         };
         Ok(GetResult {

scripts/run.sh

@@ -20,7 +20,7 @@ mkdir -p ./target/volume/test{0..4}
 if [ -z "$RUST_LOG" ]; then
     export RUST_BACKTRACE=1
 #    export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,iam=debug"
-    export RUST_LOG="rustfs=info,ecstore=info,s3s=debug"
+    export RUST_LOG="s3s=debug"
 fi
 
 # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5