diff --git a/Cargo.lock b/Cargo.lock
index 80483c573..2cc3dd5d9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1851,6 +1851,7 @@ dependencies = [
  "time",
  "tokio",
  "tokio-stream",
+ "tokio-util",
  "tonic",
  "tonic-build",
  "tonic-reflection",
@@ -2333,6 +2334,7 @@ checksum = "61e7c3654c13bcd040d4a03abee2c75b1d14a37b423cf5a813ceae1cc903ec6a"
 dependencies = [
  "bytes",
  "futures-core",
+ "futures-io",
  "futures-sink",
  "pin-project-lite",
  "tokio",
diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml
index 70d595092..79370b32f 100644
--- a/ecstore/Cargo.toml
+++ b/ecstore/Cargo.toml
@@ -39,7 +39,7 @@ nix = { version = "0.29.0", features = ["fs"] }
 path-absolutize = "3.1.1"
 protos.workspace = true
 rmp-serde = "1.3.0"
-tokio-util = { version = "0.7.12", features = ["io"] }
+tokio-util = { version = "0.7.12", features = ["io", "compat"] }
 crc32fast = "1.4.2"
 siphasher = "1.0.1"
 base64-simd = "0.8.0"
diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs
index ef5c8f9dc..98c7a6896 100644
--- a/ecstore/src/bitrot.rs
+++ b/ecstore/src/bitrot.rs
@@ -1,5 +1,5 @@
 use crate::{
-    disk::{error::DiskError, DiskStore, FileReader, FileWriter, Reader},
+    disk::{error::DiskError, DiskAPI, DiskStore, FileReader, FileWriter, Reader},
     erasure::{ReadAt, Writer},
     error::{Error, Result},
     store_api::BitrotAlgorithm,
@@ -612,7 +612,7 @@ mod test {
 
     use crate::{
         bitrot::{new_bitrot_writer, BITROT_ALGORITHMS},
-        disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption},
+        disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskAPI, DiskOption},
         error::{Error, Result},
         store_api::BitrotAlgorithm,
     };
diff --git a/ecstore/src/bucket/metadata_sys.rs b/ecstore/src/bucket/metadata_sys.rs
index 1cd4bf736..72888b1df 100644
--- a/ecstore/src/bucket/metadata_sys.rs
+++ b/ecstore/src/bucket/metadata_sys.rs
@@ -60,7 +60,7 @@ pub async fn delete(bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys().await;
     let mut bucket_meta_sys = bucket_meta_sys_lock.write().await;
 
-    bucket_meta_sys.delete(&bucket, config_file).await
+    bucket_meta_sys.delete(bucket, config_file).await
 }
 
 pub async fn get_tagging_config(bucket: &str) -> Result<(Tagging, OffsetDateTime)> {
diff --git a/ecstore/src/bucket/policy/action.rs b/ecstore/src/bucket/policy/action.rs
index cde8212bb..2b87bb1a4 100644
--- a/ecstore/src/bucket/policy/action.rs
+++ b/ecstore/src/bucket/policy/action.rs
@@ -3,6 +3,7 @@ use lazy_static::lazy_static;
 use serde::{Deserialize, Serialize};
 use std::{
     collections::{HashMap, HashSet},
+    str::FromStr,
     vec,
 };
 
@@ -282,69 +283,139 @@ impl Action {
         }
     }
 
-    pub fn from_str(s: &str) -> Option<Self> {
+    // pub fn from_str(s: &str) -> Option<Self> {
+    //     match s {
+    //         "s3:AbortMultipartUpload" => Some(Action::AbortMultipartUpload),
+    //         "s3:CreateBucket" => Some(Action::CreateBucket),
+    //         "s3:DeleteBucket" => Some(Action::DeleteBucket),
+    //         "s3:ForceDeleteBucket" => Some(Action::ForceDeleteBucket),
+    //         "s3:DeleteBucketPolicy" => Some(Action::DeleteBucketPolicy),
+    //         "s3:DeleteBucketCors" => Some(Action::DeleteBucketCors),
+    //         "s3:DeleteObject" => Some(Action::DeleteObject),
+    //         "s3:GetBucketLocation" => Some(Action::GetBucketLocation),
+    //         "s3:GetBucketNotification" => Some(Action::GetBucketNotification),
+    //         "s3:GetBucketPolicy" => Some(Action::GetBucketPolicy),
+    //         "s3:GetBucketCors" => Some(Action::GetBucketCors),
+    //         "s3:GetObject" => Some(Action::GetObject),
+    //         "s3:GetObjectAttributes" => Some(Action::GetObjectAttributes),
+    //         "s3:HeadBucket" => Some(Action::HeadBucket),
+    //         "s3:ListAllMyBuckets" => Some(Action::ListAllMyBuckets),
+    //         "s3:ListBucket" => Some(Action::ListBucket),
+    //         "s3:GetBucketPolicyStatus" => Some(Action::GetBucketPolicyStatus),
+    //         "s3:ListBucketVersions" => Some(Action::ListBucketVersions),
+    //         "s3:ListBucketMultipartUploads" => Some(Action::ListBucketMultipartUploads),
+    //         "s3:ListenNotification" => Some(Action::ListenNotification),
+    //         "s3:ListenBucketNotification" => Some(Action::ListenBucketNotification),
+    //         "s3:ListMultipartUploadParts" => Some(Action::ListMultipartUploadParts),
+    //         "s3:PutLifecycleConfiguration" => Some(Action::PutLifecycleConfiguration),
+    //         "s3:GetLifecycleConfiguration" => Some(Action::GetLifecycleConfiguration),
+    //         "s3:PutBucketNotification" => Some(Action::PutBucketNotification),
+    //         "s3:PutBucketPolicy" => Some(Action::PutBucketPolicy),
+    //         "s3:PutBucketCors" => Some(Action::PutBucketCors),
+    //         "s3:PutObject" => Some(Action::PutObject),
+    //         "s3:DeleteObjectVersion" => Some(Action::DeleteObjectVersion),
+    //         "s3:DeleteObjectVersionTagging" => Some(Action::DeleteObjectVersionTagging),
+    //         "s3:GetObjectVersion" => Some(Action::GetObjectVersion),
+    //         "s3:GetObjectVersionAttributes" => Some(Action::GetObjectVersionAttributes),
+    //         "s3:GetObjectVersionTagging" => Some(Action::GetObjectVersionTagging),
+    //         "s3:PutObjectVersionTagging" => Some(Action::PutObjectVersionTagging),
+    //         "s3:BypassGovernanceRetention" => Some(Action::BypassGovernanceRetention),
+    //         "s3:PutObjectRetention" => Some(Action::PutObjectRetention),
+    //         "s3:GetObjectRetention" => Some(Action::GetObjectRetention),
+    //         "s3:GetObjectLegalHold" => Some(Action::GetObjectLegalHold),
+    //         "s3:PutObjectLegalHold" => Some(Action::PutObjectLegalHold),
+    //         "s3:GetBucketObjectLockConfiguration" => Some(Action::GetBucketObjectLockConfiguration),
+    //         "s3:PutBucketObjectLockConfiguration" => Some(Action::PutBucketObjectLockConfiguration),
+    //         "s3:GetBucketTagging" => Some(Action::GetBucketTagging),
+    //         "s3:PutBucketTagging" => Some(Action::PutBucketTagging),
+    //         "s3:GetObjectTagging" => Some(Action::GetObjectTagging),
+    //         "s3:PutObjectTagging" => Some(Action::PutObjectTagging),
+    //         "s3:DeleteObjectTagging" => Some(Action::DeleteObjectTagging),
+    //         "s3:PutEncryptionConfiguration" => Some(Action::PutBucketEncryption),
+    //         "s3:GetEncryptionConfiguration" => Some(Action::GetBucketEncryption),
+    //         "s3:PutBucketVersioning" => Some(Action::PutBucketVersioning),
+    //         "s3:GetBucketVersioning" => Some(Action::GetBucketVersioning),
+    //         "s3:PutReplicationConfiguration" => Some(Action::PutReplicationConfiguration),
+    //         "s3:GetReplicationConfiguration" => Some(Action::GetReplicationConfiguration),
+    //         "s3:ReplicateObject" => Some(Action::ReplicateObject),
+    //         "s3:ReplicateDelete" => Some(Action::ReplicateDelete),
+    //         "s3:ReplicateTags" => Some(Action::ReplicateTags),
+    //         "s3:GetObjectVersionForReplication" => Some(Action::GetObjectVersionForReplication),
+    //         "s3:RestoreObject" => Some(Action::RestoreObject),
+    //         "s3:ResetBucketReplicationState" => Some(Action::ResetBucketReplicationState),
+    //         "s3:PutObjectFanOut" => Some(Action::PutObjectFanOut),
+    //         "s3:*" => Some(Action::AllActions),
+    //         _ => None,
+    //     }
+    // }
+}
+
+impl FromStr for Action {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
-            "s3:AbortMultipartUpload" => Some(Action::AbortMultipartUpload),
-            "s3:CreateBucket" => Some(Action::CreateBucket),
-            "s3:DeleteBucket" => Some(Action::DeleteBucket),
-            "s3:ForceDeleteBucket" => Some(Action::ForceDeleteBucket),
-            "s3:DeleteBucketPolicy" => Some(Action::DeleteBucketPolicy),
-            "s3:DeleteBucketCors" => Some(Action::DeleteBucketCors),
-            "s3:DeleteObject" => Some(Action::DeleteObject),
-            "s3:GetBucketLocation" => Some(Action::GetBucketLocation),
-            "s3:GetBucketNotification" => Some(Action::GetBucketNotification),
-            "s3:GetBucketPolicy" => Some(Action::GetBucketPolicy),
-            "s3:GetBucketCors" => Some(Action::GetBucketCors),
-            "s3:GetObject" => Some(Action::GetObject),
-            "s3:GetObjectAttributes" => Some(Action::GetObjectAttributes),
-            "s3:HeadBucket" => Some(Action::HeadBucket),
-            "s3:ListAllMyBuckets" => Some(Action::ListAllMyBuckets),
-            "s3:ListBucket" => Some(Action::ListBucket),
-            "s3:GetBucketPolicyStatus" => Some(Action::GetBucketPolicyStatus),
-            "s3:ListBucketVersions" => Some(Action::ListBucketVersions),
-            "s3:ListBucketMultipartUploads" => Some(Action::ListBucketMultipartUploads),
-            "s3:ListenNotification" => Some(Action::ListenNotification),
-            "s3:ListenBucketNotification" => Some(Action::ListenBucketNotification),
-            "s3:ListMultipartUploadParts" => Some(Action::ListMultipartUploadParts),
-            "s3:PutLifecycleConfiguration" => Some(Action::PutLifecycleConfiguration),
-            "s3:GetLifecycleConfiguration" => Some(Action::GetLifecycleConfiguration),
-            "s3:PutBucketNotification" => Some(Action::PutBucketNotification),
-            "s3:PutBucketPolicy" => Some(Action::PutBucketPolicy),
-            "s3:PutBucketCors" => Some(Action::PutBucketCors),
-            "s3:PutObject" => Some(Action::PutObject),
-            "s3:DeleteObjectVersion" => Some(Action::DeleteObjectVersion),
-            "s3:DeleteObjectVersionTagging" => Some(Action::DeleteObjectVersionTagging),
-            "s3:GetObjectVersion" => Some(Action::GetObjectVersion),
-            "s3:GetObjectVersionAttributes" => Some(Action::GetObjectVersionAttributes),
-            "s3:GetObjectVersionTagging" => Some(Action::GetObjectVersionTagging),
-            "s3:PutObjectVersionTagging" => Some(Action::PutObjectVersionTagging),
-            "s3:BypassGovernanceRetention" => Some(Action::BypassGovernanceRetention),
-            "s3:PutObjectRetention" => Some(Action::PutObjectRetention),
-            "s3:GetObjectRetention" => Some(Action::GetObjectRetention),
-            "s3:GetObjectLegalHold" => Some(Action::GetObjectLegalHold),
-            "s3:PutObjectLegalHold" => Some(Action::PutObjectLegalHold),
-            "s3:GetBucketObjectLockConfiguration" => Some(Action::GetBucketObjectLockConfiguration),
-            "s3:PutBucketObjectLockConfiguration" => Some(Action::PutBucketObjectLockConfiguration),
-            "s3:GetBucketTagging" => Some(Action::GetBucketTagging),
-            "s3:PutBucketTagging" => Some(Action::PutBucketTagging),
-            "s3:GetObjectTagging" => Some(Action::GetObjectTagging),
-            "s3:PutObjectTagging" => Some(Action::PutObjectTagging),
-            "s3:DeleteObjectTagging" => Some(Action::DeleteObjectTagging),
-            "s3:PutEncryptionConfiguration" => Some(Action::PutBucketEncryption),
-            "s3:GetEncryptionConfiguration" => Some(Action::GetBucketEncryption),
-            "s3:PutBucketVersioning" => Some(Action::PutBucketVersioning),
-            "s3:GetBucketVersioning" => Some(Action::GetBucketVersioning),
-            "s3:PutReplicationConfiguration" => Some(Action::PutReplicationConfiguration),
-            "s3:GetReplicationConfiguration" => Some(Action::GetReplicationConfiguration),
-            "s3:ReplicateObject" => Some(Action::ReplicateObject),
-            "s3:ReplicateDelete" => Some(Action::ReplicateDelete),
-            "s3:ReplicateTags" => Some(Action::ReplicateTags),
-            "s3:GetObjectVersionForReplication" => Some(Action::GetObjectVersionForReplication),
-            "s3:RestoreObject" => Some(Action::RestoreObject),
-            "s3:ResetBucketReplicationState" => Some(Action::ResetBucketReplicationState),
-            "s3:PutObjectFanOut" => Some(Action::PutObjectFanOut),
-            "s3:*" => Some(Action::AllActions),
-            _ => None,
+            "s3:AbortMultipartUpload" => Ok(Action::AbortMultipartUpload),
+            "s3:CreateBucket" => Ok(Action::CreateBucket),
+            "s3:DeleteBucket" => Ok(Action::DeleteBucket),
+            "s3:ForceDeleteBucket" => Ok(Action::ForceDeleteBucket),
+            "s3:DeleteBucketPolicy" => Ok(Action::DeleteBucketPolicy),
+            "s3:DeleteBucketCors" => Ok(Action::DeleteBucketCors),
+            "s3:DeleteObject" => Ok(Action::DeleteObject),
+            "s3:GetBucketLocation" => Ok(Action::GetBucketLocation),
+            "s3:GetBucketNotification" => Ok(Action::GetBucketNotification),
+            "s3:GetBucketPolicy" => Ok(Action::GetBucketPolicy),
+            "s3:GetBucketCors" => Ok(Action::GetBucketCors),
+            "s3:GetObject" => Ok(Action::GetObject),
+            "s3:GetObjectAttributes" => Ok(Action::GetObjectAttributes),
+            "s3:HeadBucket" => Ok(Action::HeadBucket),
+            "s3:ListAllMyBuckets" => Ok(Action::ListAllMyBuckets),
+            "s3:ListBucket" => Ok(Action::ListBucket),
+            "s3:GetBucketPolicyStatus" => Ok(Action::GetBucketPolicyStatus),
+            "s3:ListBucketVersions" => Ok(Action::ListBucketVersions),
+            "s3:ListBucketMultipartUploads" => Ok(Action::ListBucketMultipartUploads),
+            "s3:ListenNotification" => Ok(Action::ListenNotification),
+            "s3:ListenBucketNotification" => Ok(Action::ListenBucketNotification),
+            "s3:ListMultipartUploadParts" => Ok(Action::ListMultipartUploadParts),
+            "s3:PutLifecycleConfiguration" => Ok(Action::PutLifecycleConfiguration),
+            "s3:GetLifecycleConfiguration" => Ok(Action::GetLifecycleConfiguration),
+            "s3:PutBucketNotification" => Ok(Action::PutBucketNotification),
+            "s3:PutBucketPolicy" => Ok(Action::PutBucketPolicy),
+            "s3:PutBucketCors" => Ok(Action::PutBucketCors),
+            "s3:PutObject" => Ok(Action::PutObject),
+            "s3:DeleteObjectVersion" => Ok(Action::DeleteObjectVersion),
+            "s3:DeleteObjectVersionTagging" => Ok(Action::DeleteObjectVersionTagging),
+            "s3:GetObjectVersion" => Ok(Action::GetObjectVersion),
+            "s3:GetObjectVersionAttributes" => Ok(Action::GetObjectVersionAttributes),
+            "s3:GetObjectVersionTagging" => Ok(Action::GetObjectVersionTagging),
+            "s3:PutObjectVersionTagging" => Ok(Action::PutObjectVersionTagging),
+            "s3:BypassGovernanceRetention" => Ok(Action::BypassGovernanceRetention),
+            "s3:PutObjectRetention" => Ok(Action::PutObjectRetention),
+            "s3:GetObjectRetention" => Ok(Action::GetObjectRetention),
+            "s3:GetObjectLegalHold" => Ok(Action::GetObjectLegalHold),
+            "s3:PutObjectLegalHold" => Ok(Action::PutObjectLegalHold),
+            "s3:GetBucketObjectLockConfiguration" => Ok(Action::GetBucketObjectLockConfiguration),
+            "s3:PutBucketObjectLockConfiguration" => Ok(Action::PutBucketObjectLockConfiguration),
+            "s3:GetBucketTagging" => Ok(Action::GetBucketTagging),
+            "s3:PutBucketTagging" => Ok(Action::PutBucketTagging),
+            "s3:GetObjectTagging" => Ok(Action::GetObjectTagging),
+            "s3:PutObjectTagging" => Ok(Action::PutObjectTagging),
+            "s3:DeleteObjectTagging" => Ok(Action::DeleteObjectTagging),
+            "s3:PutEncryptionConfiguration" => Ok(Action::PutBucketEncryption),
+            "s3:GetEncryptionConfiguration" => Ok(Action::GetBucketEncryption),
+            "s3:PutBucketVersioning" => Ok(Action::PutBucketVersioning),
+            "s3:GetBucketVersioning" => Ok(Action::GetBucketVersioning),
+            "s3:PutReplicationConfiguration" => Ok(Action::PutReplicationConfiguration),
+            "s3:GetReplicationConfiguration" => Ok(Action::GetReplicationConfiguration),
+            "s3:ReplicateObject" => Ok(Action::ReplicateObject),
+            "s3:ReplicateDelete" => Ok(Action::ReplicateDelete),
+            "s3:ReplicateTags" => Ok(Action::ReplicateTags),
+            "s3:GetObjectVersionForReplication" => Ok(Action::GetObjectVersionForReplication),
+            "s3:RestoreObject" => Ok(Action::RestoreObject),
+            "s3:ResetBucketReplicationState" => Ok(Action::ResetBucketReplicationState),
+            "s3:PutObjectFanOut" => Ok(Action::PutObjectFanOut),
+            "s3:*" => Ok(Action::AllActions),
+            _ => Err(()),
         }
     }
 }
diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs
index 67e64c753..42b81a20f 100644
--- a/ecstore/src/cache_value/metacache_set.rs
+++ b/ecstore/src/cache_value/metacache_set.rs
@@ -10,7 +10,7 @@ use tokio::{
 };
 
 use crate::{
-    disk::{DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions},
+    disk::{DiskAPI, DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions},
     error::{Error, Result},
 };
 
diff --git a/ecstore/src/config/common.rs b/ecstore/src/config/common.rs
index cc07335bb..e5b9e3d70 100644
--- a/ecstore/src/config/common.rs
+++ b/ecstore/src/config/common.rs
@@ -173,11 +173,9 @@ async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: &ECStore, subsy
             for (i, count) in set_drive_counts.iter().enumerate() {
                 match storageclass::lookup_config(&kvs, *count) {
                     Ok(res) => {
-                        if i == 0 {
-                            if GLOBAL_StorageClass.get().is_none() {
-                                if let Err(r) = GLOBAL_StorageClass.set(res) {
-                                    error!("GLOBAL_StorageClass.set failed {:?}", r);
-                                }
+                        if i == 0 && GLOBAL_StorageClass.get().is_none() {
+                            if let Err(r) = GLOBAL_StorageClass.set(res) {
+                                error!("GLOBAL_StorageClass.set failed {:?}", r);
                             }
                         }
                     }
diff --git a/ecstore/src/config/mod.rs b/ecstore/src/config/mod.rs
index d80479cf9..7da3f8469 100644
--- a/ecstore/src/config/mod.rs
+++ b/ecstore/src/config/mod.rs
@@ -92,11 +92,9 @@ impl Config {
                     let mut default = HashMap::new();
                     default.insert("_".to_owned(), v.clone());
                     self.0.insert(k.clone(), default);
-                } else {
-                    if !self.0[k].contains_key("_") {
-                        if let Some(m) = self.0.get_mut(k) {
-                            m.insert("_".to_owned(), v.clone());
-                        }
+                } else if !self.0[k].contains_key("_") {
+                    if let Some(m) = self.0.get_mut(k) {
+                        m.insert("_".to_owned(), v.clone());
                     }
                 }
             }
diff --git a/ecstore/src/config/storageclass.rs b/ecstore/src/config/storageclass.rs
index c825875a3..411e532b7 100644
--- a/ecstore/src/config/storageclass.rs
+++ b/ecstore/src/config/storageclass.rs
@@ -307,10 +307,8 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun
         }
     }
 
-    if ss_parity > 0 && rrs_parity > 0 {
-        if ss_parity < rrs_parity {
-            return Err(Error::msg(format!("Standard storage class parity drives {} should be greater than or equal to Reduced redundancy storage class parity drives {}", ss_parity, rrs_parity)));
-        }
+    if ss_parity > 0 && rrs_parity > 0 && ss_parity < rrs_parity {
+        return Err(Error::msg(format!("Standard storage class parity drives {} should be greater than or equal to Reduced redundancy storage class parity drives {}", ss_parity, rrs_parity)));
     }
     Ok(())
 }
diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs
index 40e24c9ca..e57543058 100644
--- a/ecstore/src/disk/mod.rs
+++ b/ecstore/src/disk/mod.rs
@@ -21,9 +21,11 @@ use crate::{
 };
 use endpoint::Endpoint;
 use futures::StreamExt;
+use local::LocalDisk;
 use protos::proto_gen::node_service::{
     node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,
 };
+use remote::RemoteDisk;
 use serde::{Deserialize, Serialize};
 use std::{
     any::Any,
@@ -44,18 +46,307 @@ use tokio::{
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming};
 use tracing::info;
-use tracing::{error, warn};
+use tracing::warn;
 use uuid::Uuid;
 
-pub type DiskStore = Arc<Box<dyn DiskAPI>>;
+pub type DiskStore = Arc<Disk>;
+
+#[derive(Debug)]
+pub enum Disk {
+    Local(LocalDisk),
+    Remote(RemoteDisk),
+}
+
+#[async_trait::async_trait]
+impl DiskAPI for Disk {
+    fn to_string(&self) -> String {
+        match self {
+            Disk::Local(local_disk) => local_disk.to_string(),
+            Disk::Remote(remote_disk) => remote_disk.to_string(),
+        }
+    }
+
+    fn is_local(&self) -> bool {
+        match self {
+            Disk::Local(local_disk) => local_disk.is_local(),
+            Disk::Remote(remote_disk) => remote_disk.is_local(),
+        }
+    }
+
+    fn host_name(&self) -> String {
+        match self {
+            Disk::Local(local_disk) => local_disk.host_name(),
+            Disk::Remote(remote_disk) => remote_disk.host_name(),
+        }
+    }
+    async fn is_online(&self) -> bool {
+        match self {
+            Disk::Local(local_disk) => local_disk.is_online().await,
+            Disk::Remote(remote_disk) => remote_disk.is_online().await,
+        }
+    }
+    fn endpoint(&self) -> Endpoint {
+        match self {
+            Disk::Local(local_disk) => local_disk.endpoint(),
+            Disk::Remote(remote_disk) => remote_disk.endpoint(),
+        }
+    }
+    async fn close(&self) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.close().await,
+            Disk::Remote(remote_disk) => remote_disk.close().await,
+        }
+    }
+    fn path(&self) -> PathBuf {
+        match self {
+            Disk::Local(local_disk) => local_disk.path(),
+            Disk::Remote(remote_disk) => remote_disk.path(),
+        }
+    }
+
+    fn get_disk_location(&self) -> DiskLocation {
+        match self {
+            Disk::Local(local_disk) => local_disk.get_disk_location(),
+            Disk::Remote(remote_disk) => remote_disk.get_disk_location(),
+        }
+    }
+
+    async fn get_disk_id(&self) -> Result<Option<Uuid>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.get_disk_id().await,
+            Disk::Remote(remote_disk) => remote_disk.get_disk_id().await,
+        }
+    }
+    async fn set_disk_id(&self, id: Option<Uuid>) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.set_disk_id(id).await,
+            Disk::Remote(remote_disk) => remote_disk.set_disk_id(id).await,
+        }
+    }
+
+    async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.read_all(volume, path).await,
+            Disk::Remote(remote_disk) => remote_disk.read_all(volume, path).await,
+        }
+    }
+
+    async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await,
+            Disk::Remote(remote_disk) => remote_disk.write_all(volume, path, data).await,
+        }
+    }
+
+    async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await,
+            Disk::Remote(remote_disk) => remote_disk.delete(volume, path, opt).await,
+        }
+    }
+
+    async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
+        match self {
+            Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await,
+            Disk::Remote(remote_disk) => remote_disk.verify_file(volume, path, fi).await,
+        }
+    }
+
+    async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
+        match self {
+            Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await,
+            Disk::Remote(remote_disk) => remote_disk.check_parts(volume, path, fi).await,
+        }
+    }
+
+    async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await,
+            Disk::Remote(remote_disk) => {
+                remote_disk
+                    .rename_part(src_volume, src_path, dst_volume, dst_path, meta)
+                    .await
+            }
+        }
+    }
+    async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await,
+            Disk::Remote(remote_disk) => remote_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await,
+        }
+    }
+
+    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
+        match self {
+            Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await,
+            Disk::Remote(remote_disk) => remote_disk.create_file(_origvolume, volume, path, _file_size).await,
+        }
+    }
+
+    async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
+        match self {
+            Disk::Local(local_disk) => local_disk.append_file(volume, path).await,
+            Disk::Remote(remote_disk) => remote_disk.append_file(volume, path).await,
+        }
+    }
+
+    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,
+        }
+    }
+
+    async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, _dir_path, _count).await,
+            Disk::Remote(remote_disk) => remote_disk.list_dir(_origvolume, volume, _dir_path, _count).await,
+        }
+    }
+
+    async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.walk_dir(opts).await,
+            Disk::Remote(remote_disk) => remote_disk.walk_dir(opts).await,
+        }
+    }
+
+    async fn rename_data(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+    ) -> Result<RenameDataResp> {
+        match self {
+            Disk::Local(local_disk) => local_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await,
+            Disk::Remote(remote_disk) => remote_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await,
+        }
+    }
+
+    async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.make_volumes(volumes).await,
+            Disk::Remote(remote_disk) => remote_disk.make_volumes(volumes).await,
+        }
+    }
+
+    async fn make_volume(&self, volume: &str) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.make_volume(volume).await,
+            Disk::Remote(remote_disk) => remote_disk.make_volume(volume).await,
+        }
+    }
+
+    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.list_volumes().await,
+            Disk::Remote(remote_disk) => remote_disk.list_volumes().await,
+        }
+    }
+
+    async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
+        match self {
+            Disk::Local(local_disk) => local_disk.stat_volume(volume).await,
+            Disk::Remote(remote_disk) => remote_disk.stat_volume(volume).await,
+        }
+    }
+
+    async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await,
+            Disk::Remote(remote_disk) => remote_disk.delete_paths(volume, paths).await,
+        }
+    }
+    async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await,
+            Disk::Remote(remote_disk) => remote_disk.update_metadata(volume, path, fi, opts).await,
+        }
+    }
+
+    async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await,
+            Disk::Remote(remote_disk) => remote_disk.write_metadata(_org_volume, volume, path, fi).await,
+        }
+    }
+
+    async fn read_version(
+        &self,
+        _org_volume: &str,
+        volume: &str,
+        path: &str,
+        version_id: &str,
+        opts: &ReadOptions,
+    ) -> Result<FileInfo> {
+        match self {
+            Disk::Local(local_disk) => local_disk.read_version(_org_volume, volume, path, version_id, opts).await,
+            Disk::Remote(remote_disk) => remote_disk.read_version(_org_volume, volume, path, version_id, opts).await,
+        }
+    }
+
+    async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
+        match self {
+            Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await,
+            Disk::Remote(remote_disk) => remote_disk.read_xl(volume, path, read_data).await,
+        }
+    }
+    async fn delete_version(
+        &self,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        force_del_marker: bool,
+        opts: DeleteOptions,
+    ) -> Result<RawFileInfo> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete_version(volume, path, fi, force_del_marker, opts).await,
+            Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await,
+        }
+    }
+    async fn delete_versions(
+        &self,
+        volume: &str,
+        versions: Vec<FileInfoVersions>,
+        opts: DeleteOptions,
+    ) -> Result<Vec<Option<Error>>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete_versions(volume, versions, opts).await,
+            Disk::Remote(remote_disk) => remote_disk.delete_versions(volume, versions, opts).await,
+        }
+    }
+
+    async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
+        match self {
+            Disk::Local(local_disk) => local_disk.read_multiple(req).await,
+            Disk::Remote(remote_disk) => remote_disk.read_multiple(req).await,
+        }
+    }
+
+    async fn delete_volume(&self, volume: &str) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete_volume(volume).await,
+            Disk::Remote(remote_disk) => remote_disk.delete_volume(volume).await,
+        }
+    }
+
+    async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
+        match self {
+            Disk::Local(local_disk) => local_disk.disk_info(opts).await,
+            Disk::Remote(remote_disk) => remote_disk.disk_info(opts).await,
+        }
+    }
+}
 
 pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result<DiskStore> {
     if ep.is_local {
         let s = local::LocalDisk::new(ep, opt.cleanup).await?;
-        Ok(Arc::new(Box::new(s)))
+        Ok(Arc::new(Disk::Local(s)))
     } else {
         let remote_disk = remote::RemoteDisk::new(ep, opt).await?;
-        Ok(Arc::new(Box::new(remote_disk)))
+        Ok(Arc::new(Disk::Remote(remote_disk)))
     }
 }
 
diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs
index 35ecb2345..0f4baa4f7 100644
--- a/ecstore/src/erasure.rs
+++ b/ecstore/src/erasure.rs
@@ -1,4 +1,4 @@
-use crate::bitrot::{close_bitrot_writers, BitrotReader, BitrotWriter};
+use crate::bitrot::{BitrotReader, BitrotWriter};
 use crate::error::{Error, Result, StdError};
 use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs};
 use bytes::Bytes;
@@ -7,15 +7,16 @@ use futures::{pin_mut, Stream, StreamExt};
 use reed_solomon_erasure::galois_8::ReedSolomon;
 use std::any::Any;
 use std::fmt::Debug;
-use tokio::io::AsyncWriteExt;
+use std::io::ErrorKind;
 use tokio::io::DuplexStream;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tracing::debug;
 use tracing::warn;
 // use tracing::debug;
 use uuid::Uuid;
 
 // use reader::reader::ChunkedStream;
-use crate::chunk_stream::ChunkedStream;
+// use crate::chunk_stream::ChunkedStream;
 use crate::disk::error::DiskError;
 
 pub struct Erasure {
@@ -24,6 +25,7 @@ pub struct Erasure {
     encoder: Option<ReedSolomon>,
     pub block_size: usize,
     _id: Uuid,
+    buf: Vec<u8>,
 }
 
 impl Erasure {
@@ -43,12 +45,13 @@ impl Erasure {
             block_size,
             encoder,
             _id: Uuid::new_v4(),
+            buf: vec![0u8; block_size],
         }
     }
 
     #[tracing::instrument(level = "debug", skip(self, body, writers))]
     pub async fn encode<S>(
-        &self,
+        &mut self,
         body: S,
         writers: &mut [Option<BitrotWriter>],
         // block_size: usize,
@@ -58,76 +61,138 @@ impl Erasure {
     where
         S: Stream<Item = Result<Bytes, StdError>> + Send + Sync,
     {
-        // let stream = ChunkedStream::new(body, self.block_size);
-        let stream = ChunkedStream::new(body, total_size, self.block_size, false);
+        pin_mut!(body);
+        let mut reader = tokio_util::io::StreamReader::new(
+            body.map(|f| f.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))),
+        );
+
         let mut total: usize = 0;
-        // let mut idx = 0;
-        pin_mut!(stream);
-
-        // warn!("encode start...");
-
         loop {
-            match stream.next().await {
-                Some(result) => match result {
-                    Ok(data) => {
-                        total += data.len();
-
-                        // EOF
-                        if data.is_empty() {
-                            break;
-                        }
-
-                        // idx += 1;
-                        // warn!("encode {} get data {:?}", data.len(), data.to_vec());
-
-                        let blocks = self.encode_data(data.as_ref())?;
-
-                        // warn!(
-                        //     "encode shard  size: {}/{} from block_size {}, total_size {} ",
-                        //     blocks[0].len(),
-                        //     blocks.len(),
-                        //     data.len(),
-                        //     total_size
-                        // );
-
-                        let mut errs = Vec::new();
-
-                        for (i, w_op) in writers.iter_mut().enumerate() {
-                            if let Some(w) = w_op {
-                                match w.write(blocks[i].as_ref()).await {
-                                    Ok(_) => errs.push(None),
-                                    Err(e) => errs.push(Some(e)),
-                                }
-                            } else {
-                                errs.push(Some(Error::new(DiskError::DiskNotFound)));
-                            }
-                        }
-
-                        let none_count = errs.iter().filter(|&x| x.is_none()).count();
-                        if none_count >= write_quorum {
-                            continue;
-                        }
-
-                        if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
-                            warn!("Erasure encode errs {:?}", &errs);
-                            return Err(err);
-                        }
-                    }
-                    Err(e) => {
-                        warn!("poll result err {:?}", &e);
-                        return Err(Error::msg(e.to_string()));
-                    }
-                },
-                None => {
-                    // warn!("poll empty result");
-                    break;
+            let new_len = {
+                let remain = total_size - total;
+                if remain > self.block_size {
+                    self.block_size
+                } else {
+                    remain
                 }
+            };
+
+            if new_len == 0 {
+                break;
+            }
+
+            self.buf.resize(new_len, 0u8);
+
+            match reader.read_exact(&mut self.buf).await {
+                Ok(res) => res,
+                Err(e) => {
+                    if let ErrorKind::UnexpectedEof = e.kind() {
+                        break;
+                    } else {
+                        return Err(Error::new(e));
+                    }
+                }
+            };
+
+            total += self.buf.len();
+
+            let blocks = self.encode_data(&self.buf)?;
+            let mut errs = Vec::new();
+
+            for (i, w_op) in writers.iter_mut().enumerate() {
+                if let Some(w) = w_op {
+                    match w.write(blocks[i].as_ref()).await {
+                        Ok(_) => errs.push(None),
+                        Err(e) => errs.push(Some(e)),
+                    }
+                } else {
+                    errs.push(Some(Error::new(DiskError::DiskNotFound)));
+                }
+            }
+
+            let none_count = errs.iter().filter(|&x| x.is_none()).count();
+            if none_count >= write_quorum {
+                continue;
+            }
+
+            if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
+                warn!("Erasure encode errs {:?}", &errs);
+                return Err(err);
             }
         }
 
-        let _ = close_bitrot_writers(writers).await?;
-
         Ok(total)
+
+        // // let stream = ChunkedStream::new(body, self.block_size);
+        // let stream = ChunkedStream::new(body, total_size, self.block_size, false);
+        // let mut total: usize = 0;
+        // // let mut idx = 0;
+        // pin_mut!(stream);
+
+        // // warn!("encode start...");
+
+        // loop {
+        //     match stream.next().await {
+        //         Some(result) => match result {
+        //             Ok(data) => {
+        //                 total += data.len();
+
+        //                 // EOF
+        //                 if data.is_empty() {
+        //                     break;
+        //                 }
+
+        //                 // idx += 1;
+        //                 // warn!("encode {} get data {:?}", data.len(), data.to_vec());
+
+        //                 let blocks = self.encode_data(data.as_ref())?;
+
+        //                 // warn!(
+        //                 //     "encode shard  size: {}/{} from block_size {}, total_size {} ",
+        //                 //     blocks[0].len(),
+        //                 //     blocks.len(),
+        //                 //     data.len(),
+        //                 //     total_size
+        //                 // );
+
+        //                 let mut errs = Vec::new();
+
+        //                 for (i, w_op) in writers.iter_mut().enumerate() {
+        //                     if let Some(w) = w_op {
+        //                         match w.write(blocks[i].as_ref()).await {
+        //                             Ok(_) => errs.push(None),
+        //                             Err(e) => errs.push(Some(e)),
+        //                         }
+        //                     } else {
+        //                         errs.push(Some(Error::new(DiskError::DiskNotFound)));
+        //                     }
+        //                 }
+
+        //                 let none_count = errs.iter().filter(|&x| x.is_none()).count();
+        //                 if none_count >= write_quorum {
+        //                     continue;
+        //                 }
+
+        //                 if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
+        //                     warn!("Erasure encode errs {:?}", &errs);
+        //                     return Err(err);
+        //                 }
+        //             }
+        //             Err(e) => {
+        //                 warn!("poll result err {:?}", &e);
+        //                 return Err(Error::msg(e.to_string()));
+        //             }
+        //         },
+        //         None => {
+        //             // warn!("poll empty result");
+        //             break;
+        //         }
+        //     }
+        // }
+
+        // let _ = close_bitrot_writers(writers).await?;
+
+        // Ok(total)
     }
 
     pub async fn decode(
diff --git a/ecstore/src/file_meta_inline.rs b/ecstore/src/file_meta_inline.rs
index 0ec834239..a85813553 100644
--- a/ecstore/src/file_meta_inline.rs
+++ b/ecstore/src/file_meta_inline.rs
@@ -16,7 +16,7 @@ impl InlineData {
         self.0 = buf.to_vec()
     }
     pub fn as_slice(&self) -> &[u8] {
-        &self.0.as_slice()
+        self.0.as_slice()
     }
     pub fn version_ok(&self) -> bool {
         if self.0.is_empty() {
diff --git a/ecstore/src/heal/heal_commands.rs b/ecstore/src/heal/heal_commands.rs
index cc2578041..39da448e2 100644
--- a/ecstore/src/heal/heal_commands.rs
+++ b/ecstore/src/heal/heal_commands.rs
@@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::RwLock;
 
 use crate::{
-    disk::{DeleteOptions, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
+    disk::{DeleteOptions, DiskAPI, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
     error::{Error, Result},
     heal::heal_ops::HEALING_TRACKER_FILENAME,
     new_object_layer_fn,
diff --git a/ecstore/src/heal/heal_ops.rs b/ecstore/src/heal/heal_ops.rs
index ae7acd6db..9ef9ee37a 100644
--- a/ecstore/src/heal/heal_ops.rs
+++ b/ecstore/src/heal/heal_ops.rs
@@ -325,12 +325,12 @@ pub async fn heal_sequence_start(h: Arc<HealSequence>) {
                     match err {
                         Some(err) => {
                             let mut current_status_w = h.current_status.write().await;
-                            (*current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
-                            (*current_status_w).failure_detail = err.to_string();
+                            (current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
+                            (current_status_w).failure_detail = err.to_string();
                         },
                         None => {
                             let mut current_status_w = h.current_status.write().await;
-                            (*current_status_w).summary = HEAL_FINISHED_STATUS.to_string();
+                            (current_status_w).summary = HEAL_FINISHED_STATUS.to_string();
                         }
                     }
                 },
@@ -529,7 +529,7 @@ impl AllHealState {
         let _ = self.mu.write().await;
 
         for (k, v) in self.heal_seq_map.iter() {
-            if !v.has_ended().await && (has_profix(&k, path_s) || has_profix(path_s, &k)) {
+            if !v.has_ended().await && (has_profix(k, path_s) || has_profix(path_s, k)) {
                 return Err(Error::from_string(format!(
                     "The provided heal sequence path overlaps with an existing heal path: {}",
                     k
@@ -546,7 +546,6 @@ impl AllHealState {
 
         if heal_sequence.client_token == BG_HEALING_UUID {
             // For background heal do nothing, do not spawn an unnecessary goroutine.
-        } else {
         }
         todo!()
     }
diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs
index 35e0d1b01..a56fdb426 100644
--- a/ecstore/src/peer.rs
+++ b/ecstore/src/peer.rs
@@ -7,6 +7,7 @@ use std::{collections::HashMap, fmt::Debug, sync::Arc};
 use tonic::Request;
 use tracing::warn;
 
+use crate::disk::DiskAPI;
 use crate::store::all_local_disk;
 use crate::{
     disk::{self, error::DiskError, VolumeInfo},
diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs
index 74068768e..c5dc2c07c 100644
--- a/ecstore/src/sets.rs
+++ b/ecstore/src/sets.rs
@@ -11,7 +11,7 @@ use uuid::Uuid;
 use crate::{
     disk::{
         format::{DistributionAlgoVersion, FormatV3},
-        DiskStore,
+        DiskAPI, DiskStore,
     },
     endpoints::PoolEndpoints,
     error::{Error, Result},
@@ -454,7 +454,7 @@ impl StorageAPI for Sets {
         delimiter: &str,
         max_uploads: usize,
     ) -> Result<ListMultipartsInfo> {
-        self.get_disks_by_key(&prefix)
+        self.get_disks_by_key(prefix)
             .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads)
             .await
     }
diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs
index 96ed9a106..98627c709 100644
--- a/ecstore/src/store.rs
+++ b/ecstore/src/store.rs
@@ -5,7 +5,7 @@ use crate::bucket::metadata_sys::{self, init_bucket_metadata_sys, set_bucket_met
 use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
 use crate::config::{self, storageclass, GLOBAL_ConfigSys};
 use crate::disk::endpoint::EndpointType;
-use crate::disk::{DiskInfo, DiskInfoOptions, MetaCacheEntry};
+use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions, MetaCacheEntry};
 use crate::global::{
     is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION,
     DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES,
@@ -331,7 +331,7 @@ impl ECStore {
 
     async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {
         let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await;
-        server_pools.filter_max_used(100 - (100 as f64 * DISK_RESERVE_FRACTION) as u64);
+        server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64);
         let total = server_pools.total_available();
 
         if total == 0 {
@@ -382,10 +382,7 @@ impl ECStore {
             }
 
             if !is_meta_bucketname(bucket) {
-                let avail = match has_space_for(zinfo, size).await {
-                    Ok(res) => res,
-                    Err(_err) => false,
-                };
+                let avail = has_space_for(zinfo, size).await.unwrap_or_default();
 
                 if !avail {
                     server_pools.push(PoolAvailableSpace {
@@ -399,19 +396,17 @@ impl ECStore {
 
             let mut available = 0;
             let mut max_used_pct = 0;
-            for disk_op in zinfo.iter() {
-                if let Some(disk) = disk_op {
-                    if disk.total == 0 {
-                        continue;
-                    }
+            for disk in zinfo.iter().flatten() {
+                if disk.total == 0 {
+                    continue;
+                }
 
-                    available += disk.total - disk.used;
+                available += disk.total - disk.used;
 
-                    let pct_used = disk.used * 100 / disk.total;
+                let pct_used = disk.used * 100 / disk.total;
 
-                    if pct_used > max_used_pct {
-                        max_used_pct = pct_used;
-                    }
+                if pct_used > max_used_pct {
+                    max_used_pct = pct_used;
                 }
             }
 
@@ -769,7 +764,7 @@ impl ObjectIO for ECStore {
             return self.pools[0].put_object(bucket, object.as_str(), data, opts).await;
         }
 
-        let idx = self.get_pool_idx(&bucket, &object, data.content_length as i64).await?;
+        let idx = self.get_pool_idx(bucket, &object, data.content_length as i64).await?;
 
         if opts.data_movement && idx == opts.src_pool_idx {
             return Err(Error::new(StorageError::DataMovementOverwriteErr(
@@ -812,11 +807,11 @@ impl StorageAPI for ECStore {
     }
 
     async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
-        if is_meta_bucketname(&bucket) {
+        if is_meta_bucketname(bucket) {
             return Err(StorageError::BucketNameInvalid(bucket.to_string()).into());
         }
 
-        if let Err(err) = check_valid_bucket_name(&bucket) {
+        if let Err(err) = check_valid_bucket_name(bucket) {
             return Err(StorageError::BucketNameInvalid(err.to_string()).into());
         }
 
@@ -838,8 +833,8 @@ impl StorageAPI for ECStore {
         Ok(())
     }
     async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
-        if !is_meta_bucketname(&bucket) {
-            if let Err(err) = check_valid_bucket_name_strict(&bucket) {
+        if !is_meta_bucketname(bucket) {
+            if let Err(err) = check_valid_bucket_name_strict(bucket) {
                 return Err(StorageError::BucketNameInvalid(err.to_string()).into());
             }
 
@@ -885,7 +880,7 @@ impl StorageAPI for ECStore {
     async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
         let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?;
 
-        if let Ok(sys) = metadata_sys::get(&bucket).await {
+        if let Ok(sys) = metadata_sys::get(bucket).await {
             info.created = Some(sys.created);
             info.versionning = sys.versioning();
             info.object_locking = sys.object_locking();
@@ -1234,7 +1229,7 @@ impl StorageAPI for ECStore {
         for (idx, pool) in self.pools.iter().enumerate() {
             // // TODO: IsSuspended
             let res = pool
-                .list_multipart_uploads(bucket, &object, "", "", "", MAX_UPLOADS_LIST)
+                .list_multipart_uploads(bucket, object, "", "", "", MAX_UPLOADS_LIST)
                 .await?;
 
             if !res.uploads.is_empty() {
@@ -1376,7 +1371,7 @@ impl StorageAPI for ECStore {
 
         // Return the first nil error
         for i in 0..self.pools.len() {
-            if errs.get(&i).is_none() {
+            if !errs.contains_key(&i) {
                 return Ok(results.remove(&i).unwrap());
             }
         }
@@ -1385,7 +1380,7 @@ impl StorageAPI for ECStore {
         for (k, err) in errs.iter() {
             match err.downcast_ref::<DiskError>() {
                 Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
-                _ => return Ok(results.remove(&k).unwrap()),
+                _ => return Ok(results.remove(k).unwrap()),
             }
         }
 
@@ -1611,10 +1606,7 @@ fn check_multipart_object_args(bucket: &str, object: &str, upload_id: &str) -> R
     if let Err(e) = base64_decode(upload_id.as_bytes()) {
         return Err(Error::new(StorageError::MalformedUploadID(format!(
             "{}/{}-{},err:{}",
-            bucket,
-            object,
-            upload_id,
-            e.to_string()
+            bucket, object, upload_id, e
         ))));
     };
     check_object_args(bucket, object)
@@ -1711,7 +1703,7 @@ pub async fn heal_entry(
     Ok(())
 }
 
-async fn get_disk_infos(disks: &Vec<Option<DiskStore>>) -> Vec<Option<DiskInfo>> {
+async fn get_disk_infos(disks: &[Option<DiskStore>]) -> Vec<Option<DiskInfo>> {
     let opts = &DiskInfoOptions::default();
     let mut res = vec![None; disks.len()];
     for (idx, disk_op) in disks.iter().enumerate() {
@@ -1776,7 +1768,7 @@ impl ServerPoolsAvailableSpace {
     }
 }
 
-async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> {
+async fn has_space_for(dis: &[Option<DiskInfo>], size: i64) -> Result<bool> {
     let size = {
         if size < 0 {
             DISK_ASSUME_UNKNOWN_SIZE
@@ -1789,12 +1781,10 @@ async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> {
     let mut total = 0;
     let mut disks_num = 0;
 
-    for disk_op in dis.iter() {
-        if let Some(disk) = disk_op {
-            disks_num += 1;
-            total += disk.total;
-            available += disk.total - disk.used;
-        }
+    for disk in dis.iter().flatten() {
+        disks_num += 1;
+        total += disk.total;
+        available += disk.total - disk.used;
     }
 
     if disks_num < dis.len() / 2 || disks_num == 0 {
@@ -1807,15 +1797,13 @@ async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> {
 
     let per_disk = size / disks_num as u64;
 
-    for disk_op in dis.iter() {
-        if let Some(disk) = disk_op {
-            if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 {
-                return Ok(false);
-            }
+    for disk in dis.iter().flatten() {
+        if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 {
+            return Ok(false);
+        }
 
-            if disk.free <= per_disk {
-                return Ok(false);
-            }
+        if disk.free <= per_disk {
+            return Ok(false);
         }
     }
 
diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs
index 7afc84efc..21df384ed 100644
--- a/ecstore/src/store_api.rs
+++ b/ecstore/src/store_api.rs
@@ -172,27 +172,9 @@ impl FileInfo {
 
         let (content_type, content_encoding, etag) = {
             if let Some(ref meta) = self.metadata {
-                let content_type = {
-                    if let Some(ty) = meta.get("content-type") {
-                        Some(ty.clone())
-                    } else {
-                        None
-                    }
-                };
-                let content_encoding = {
-                    if let Some(encoding) = meta.get("content-encoding") {
-                        Some(encoding.clone())
-                    } else {
-                        None
-                    }
-                };
-                let etag = {
-                    if let Some(etag) = meta.get("etag") {
-                        Some(etag.clone())
-                    } else {
-                        None
-                    }
-                };
+                let content_type = meta.get("content-type").cloned();
+                let content_encoding = meta.get("content-encoding").cloned();
+                let etag = meta.get("etag").cloned();
 
                 (content_type, content_encoding, etag)
             } else {
diff --git a/ecstore/src/store_err.rs b/ecstore/src/store_err.rs
index 072ff46d6..d8c468d12 100644
--- a/ecstore/src/store_err.rs
+++ b/ecstore/src/store_err.rs
@@ -42,10 +42,7 @@ pub enum StorageError {
 
 pub fn is_err_invalid_upload_id(err: &Error) -> bool {
     if let Some(e) = err.downcast_ref::<StorageError>() {
-        match e {
-            StorageError::InvalidUploadID(_, _, _) => true,
-            _ => false,
-        }
+        matches!(e, StorageError::InvalidUploadID(_, _, _))
     } else {
         false
     }
@@ -53,10 +50,7 @@ pub fn is_err_invalid_upload_id(err: &Error) -> bool {
 
 pub fn is_err_version_not_found(err: &Error) -> bool {
     if let Some(e) = err.downcast_ref::<StorageError>() {
-        match e {
-            StorageError::VersionNotFound(_, _, _) => true,
-            _ => false,
-        }
+        matches!(e, StorageError::VersionNotFound(_, _, _))
     } else {
         false
     }
@@ -64,10 +58,7 @@ pub fn is_err_version_not_found(err: &Error) -> bool {
 
 pub fn is_err_bucket_exists(err: &Error) -> bool {
     if let Some(e) = err.downcast_ref::<StorageError>() {
-        match e {
-            StorageError::BucketExists(_) => true,
-            _ => false,
-        }
+        matches!(e, StorageError::BucketExists(_))
     } else {
         false
     }
@@ -78,10 +69,7 @@ pub fn is_err_object_not_found(err: &Error) -> bool {
         return true;
     }
     if let Some(e) = err.downcast_ref::<StorageError>() {
-        match e {
-            StorageError::ObjectNotFound(_, _) => true,
-            _ => false,
-        }
+        matches!(e, StorageError::ObjectNotFound(_, _))
     } else {
         false
     }
diff --git a/ecstore/src/store_init.rs b/ecstore/src/store_init.rs
index f1ce92dff..6d8f7dd53 100644
--- a/ecstore/src/store_init.rs
+++ b/ecstore/src/store_init.rs
@@ -1,4 +1,5 @@
 use crate::config::{storageclass, KVS};
+use crate::disk::DiskAPI;
 use crate::{
     disk::{
         error::DiskError,
diff --git a/reader/src/hasher.rs b/reader/src/hasher.rs
index 80cc51d88..37f28509b 100644
--- a/reader/src/hasher.rs
+++ b/reader/src/hasher.rs
@@ -78,6 +78,11 @@ impl Sha256 {
         }
     }
 }
+impl Default for Sha256 {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 
 impl Hasher for Sha256 {
     fn write(&mut self, bytes: &[u8]) {
@@ -110,6 +115,11 @@ impl MD5 {
         Self { hasher: Md5::new() }
     }
 }
+impl Default for MD5 {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 
 impl Hasher for MD5 {
     fn write(&mut self, bytes: &[u8]) {
@@ -142,9 +152,7 @@ impl Uuid {
 }
 
 impl Hasher for Uuid {
-    fn write(&mut self, _bytes: &[u8]) {
-        ()
-    }
+    fn write(&mut self, _bytes: &[u8]) {}
 
     fn reset(&mut self) {}
 
diff --git a/reader/src/reader.rs b/reader/src/reader.rs
index d5808c8c7..1758036c1 100644
--- a/reader/src/reader.rs
+++ b/reader/src/reader.rs
@@ -1,7 +1,6 @@
 use bytes::Bytes;
 use s3s::StdError;
 use std::collections::VecDeque;
-use tracing::warn;
 
 use std::pin::Pin;
 use std::task::Poll;
@@ -145,7 +144,7 @@ where
                     }
 
                     if let Some(md5) = this.md5 {
-                        let _ = md5.write(bytes);
+                        md5.write(bytes);
                     }
                 }
                 Some(Err(err)) => {
@@ -231,7 +230,7 @@ where
 
                         // println!("get len {}", bytes.len());
                         // 如果有剩余
-                        if this.remaining.len() > 0 {
+                        if !this.remaining.is_empty() {
                             let need_size = chuck_size - this.remaining.len();
                             // 传入的数据大小需要补齐的大小，使用传入数据补齐
                             if bytes.len() >= need_size {
@@ -255,7 +254,7 @@ where
                             this.streams.push_back(chuck);
                         }
 
-                        if bytes.len() > 0 {
+                        if !bytes.is_empty() {
                             this.remaining.extend_from_slice(&bytes);
                         }
 
@@ -267,21 +266,21 @@ where
                             return Poll::Pending;
                         }
 
-                        if this.remaining.len() > 0 {
+                        if !this.remaining.is_empty() {
                             let b = this.remaining.clone();
                             this.remaining.clear();
                             return Poll::Ready(Some(Ok(Bytes::from(b))));
                         }
                         Poll::Ready(None)
                     }
-                    Err(err) => return Poll::Ready(Some(Err(err))),
+                    Err(err) => Poll::Ready(Some(Err(err))),
                 },
                 None => {
                     // println!("get empty");
                     if let Some(b) = this.streams.pop_front() {
                         return Poll::Ready(Some(Ok(b)));
                     }
-                    if this.remaining.len() > 0 {
+                    if !this.remaining.is_empty() {
                         let b = this.remaining.clone();
                         this.remaining.clear();
                         return Poll::Ready(Some(Ok(Bytes::from(b))));
@@ -343,7 +342,7 @@ where
 
     fn size_hint(&self) -> (usize, Option<usize>) {
         let mut items = self.streams.len();
-        if self.remaining.len() > 0 {
+        if !self.remaining.is_empty() {
             items += 1;
         }
         (items, Some(items))
diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml
index 424132dfe..deb2ed322 100644
--- a/rustfs/Cargo.toml
+++ b/rustfs/Cargo.toml
@@ -35,6 +35,7 @@ s3s.workspace = true
 serde_json.workspace = true
 tracing.workspace = true
 time = { workspace = true, features = ["parsing", "formatting"] }
+tokio-util = { version = "0.7.12", features = ["io", "compat"] }
 tokio = { workspace = true, features = [
     "rt-multi-thread",
     "macros",
diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs
index 9f3577715..4e8b00636 100644
--- a/rustfs/src/grpc.rs
+++ b/rustfs/src/grpc.rs
@@ -2,8 +2,8 @@ use std::{error::Error, io::ErrorKind, pin::Pin};
 
 use ecstore::{
     disk::{
-        DeleteOptions, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, Reader, UpdateMetadataOpts,
-        WalkDirOptions,
+        DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, Reader,
+        UpdateMetadataOpts, WalkDirOptions,
     },
     erasure::{ReadAt, Writer},
     peer::{LocalPeerS3Client, PeerS3Client},
diff --git a/rustfs/src/storage/acess.rs b/rustfs/src/storage/acess.rs
index 560d5a491..b437a3594 100644
--- a/rustfs/src/storage/acess.rs
+++ b/rustfs/src/storage/acess.rs
@@ -1,3 +1,5 @@
+use std::str::FromStr;
+
 use super::ecfs::FS;
 
 use ecstore::bucket::policy::action::Action;
@@ -42,7 +44,10 @@ impl S3Access for FS {
         // 上层验证了 ak/sk
         // warn!("check s3_op {:?} cred {:?}", cx.s3_op().name(), cx.credentials());
 
-        let action = Action::from_str(format!("s3:{}", cx.s3_op().name()).as_str());
+        let action = match Action::from_str(format!("s3:{}", cx.s3_op().name()).as_str()) {
+            Ok(res) => Some(res),
+            Err(_) => None,
+        };
 
         let req_info = ReqInfo {
             card: cx.credentials().cloned(),
diff --git a/scripts/run.sh b/scripts/run.sh
index fce58dc4c..371bd4113 100755
--- a/scripts/run.sh
+++ b/scripts/run.sh
@@ -6,9 +6,9 @@ mkdir -p ./target/volume/test
 mkdir -p ./target/volume/test{0..4}
 
 
-if [ -z "$RUST_LOG" ]; then
-    export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,reader=debug"
-fi
+# if [ -z "$RUST_LOG" ]; then
+#     export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,reader=debug"
+# fi
 
 # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5