Merge pull request #480 from rustfs/feat/compress

feat: add object compression support

ecstore/src/bitrot.rs

@@ -68,14 +68,20 @@ pub async fn create_bitrot_writer(
     disk: Option<&DiskStore>,
     volume: &str,
     path: &str,
-    length: usize,
+    length: i64,
     shard_size: usize,
     checksum_algo: HashAlgorithm,
 ) -> disk::error::Result<BitrotWriterWrapper> {
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = length.div_ceil(shard_size) * checksum_algo.size() + length;
+        let length = if length > 0 {
+            let length = length as usize;
+            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+        } else {
+            0
+        };
+
         let file = disk.create_file("", volume, path, length).await?;
         CustomWriter::new_tokio_writer(file)
     } else {

ecstore/src/bucket/metadata_sys.rs

@@ -443,7 +443,6 @@ impl BucketMetadataSys {
         let bm = match self.get_config(bucket).await {
             Ok((res, _)) => res,
             Err(err) => {
-                warn!("get_object_lock_config err {:?}", &err);
                 return if err == Error::ConfigNotFound {
                     Err(BucketMetadataError::BucketObjectLockConfigNotFound.into())
                 } else {

ecstore/src/cmd/bucket_replication.rs

@@ -511,8 +511,8 @@ pub async fn get_heal_replicate_object_info(
 
     let mut result = ReplicateObjectInfo {
         name: oi.name.clone(),
-        size: oi.size as i64,
-        actual_size: asz as i64,
+        size: oi.size,
+        actual_size: asz,
         bucket: oi.bucket.clone(),
         //version_id: oi.version_id.clone(),
         version_id: oi
@@ -814,8 +814,8 @@ impl ReplicationPool {
             vsender.pop(); // Dropping the sender will close the channel
         }
         self.workers_sender = vsender;
-        warn!("self sender size is {:?}", self.workers_sender.len());
-        warn!("self sender size is {:?}", self.workers_sender.len());
+        // warn!("self sender size is {:?}", self.workers_sender.len());
+        // warn!("self sender size is {:?}", self.workers_sender.len());
     }
 
     async fn resize_failed_workers(&self, _count: usize) {
@@ -1758,13 +1758,13 @@ pub async fn schedule_replication(oi: ObjectInfo, o: Arc<store::ECStore>, dsc: R
     let replication_timestamp = Utc::now(); // Placeholder for timestamp parsing
     let replication_state = oi.replication_state();
 
-    let actual_size = oi.actual_size.unwrap_or(0);
+    let actual_size = oi.actual_size;
     //let ssec = oi.user_defined.contains_key("ssec");
     let ssec = false;
 
     let ri = ReplicateObjectInfo {
         name: oi.name,
-        size: oi.size as i64,
+        size: oi.size,
         bucket: oi.bucket,
         version_id: oi
             .version_id
@@ -2018,8 +2018,8 @@ impl ReplicateObjectInfo {
             mod_time: Some(
                 OffsetDateTime::from_unix_timestamp(self.mod_time.timestamp()).unwrap_or_else(|_| OffsetDateTime::now_utc()),
             ),
-            size: self.size as usize,
-            actual_size: Some(self.actual_size as usize),
+            size: self.size,
+            actual_size: self.actual_size,
             is_dir: false,
             user_defined: None, // 可以按需从别处导入
             parity_blocks: 0,
@@ -2317,7 +2317,7 @@ impl ReplicateObjectInfo {
 
         // 设置对象大小
         //rinfo.size = object_info.actual_size.unwrap_or(0);
-        rinfo.size = object_info.actual_size.map_or(0, |v| v as i64);
+        rinfo.size = object_info.actual_size;
         //rinfo.replication_action = object_info.
 
         rinfo.replication_status = ReplicationStatusType::Completed;

ecstore/src/compress.rs

@@ -0,0 +1,115 @@
+use rustfs_utils::string::has_pattern;
+use rustfs_utils::string::has_string_suffix_in_slice;
+use std::env;
+use tracing::error;
+
+pub const MIN_COMPRESSIBLE_SIZE: usize = 4096;
+
+// 环境变量名称，用于控制是否启用压缩
+pub const ENV_COMPRESSION_ENABLED: &str = "RUSTFS_COMPRESSION_ENABLED";
+
+// Some standard object extensions which we strictly dis-allow for compression.
+pub const STANDARD_EXCLUDE_COMPRESS_EXTENSIONS: &[&str] = &[
+    ".gz", ".bz2", ".rar", ".zip", ".7z", ".xz", ".mp4", ".mkv", ".mov", ".jpg", ".png", ".gif",
+];
+
+// Some standard content-types which we strictly dis-allow for compression.
+pub const STANDARD_EXCLUDE_COMPRESS_CONTENT_TYPES: &[&str] = &[
+    "video/*",
+    "audio/*",
+    "application/zip",
+    "application/x-gzip",
+    "application/x-zip-compressed",
+    "application/x-compress",
+    "application/x-spoon",
+];
+
+pub fn is_compressible(headers: &http::HeaderMap, object_name: &str) -> bool {
+    // 检查环境变量是否启用压缩，默认关闭
+    if let Ok(compression_enabled) = env::var(ENV_COMPRESSION_ENABLED) {
+        if compression_enabled.to_lowercase() != "true" {
+            error!("Compression is disabled by environment variable");
+            return false;
+        }
+    } else {
+        // 环境变量未设置时默认关闭
+        return false;
+    }
+
+    let content_type = headers.get("content-type").and_then(|s| s.to_str().ok()).unwrap_or("");
+
+    // TODO: crypto request return false
+
+    if has_string_suffix_in_slice(object_name, STANDARD_EXCLUDE_COMPRESS_EXTENSIONS) {
+        error!("object_name: {} is not compressible", object_name);
+        return false;
+    }
+
+    if !content_type.is_empty() && has_pattern(STANDARD_EXCLUDE_COMPRESS_CONTENT_TYPES, content_type) {
+        error!("content_type: {} is not compressible", content_type);
+        return false;
+    }
+    true
+
+    // TODO: check from config
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use temp_env;
+
+    #[test]
+    fn test_is_compressible() {
+        use http::HeaderMap;
+
+        let headers = HeaderMap::new();
+
+        // 测试环境变量控制
+        temp_env::with_var(ENV_COMPRESSION_ENABLED, Some("false"), || {
+            assert!(!is_compressible(&headers, "file.txt"));
+        });
+
+        temp_env::with_var(ENV_COMPRESSION_ENABLED, Some("true"), || {
+            assert!(is_compressible(&headers, "file.txt"));
+        });
+
+        temp_env::with_var_unset(ENV_COMPRESSION_ENABLED, || {
+            assert!(!is_compressible(&headers, "file.txt"));
+        });
+
+        temp_env::with_var(ENV_COMPRESSION_ENABLED, Some("true"), || {
+            let mut headers = HeaderMap::new();
+            // 测试不可压缩的扩展名
+            headers.insert("content-type", "text/plain".parse().unwrap());
+            assert!(!is_compressible(&headers, "file.gz"));
+            assert!(!is_compressible(&headers, "file.zip"));
+            assert!(!is_compressible(&headers, "file.mp4"));
+            assert!(!is_compressible(&headers, "file.jpg"));
+
+            // 测试不可压缩的内容类型
+            headers.insert("content-type", "video/mp4".parse().unwrap());
+            assert!(!is_compressible(&headers, "file.txt"));
+
+            headers.insert("content-type", "audio/mpeg".parse().unwrap());
+            assert!(!is_compressible(&headers, "file.txt"));
+
+            headers.insert("content-type", "application/zip".parse().unwrap());
+            assert!(!is_compressible(&headers, "file.txt"));
+
+            headers.insert("content-type", "application/x-gzip".parse().unwrap());
+            assert!(!is_compressible(&headers, "file.txt"));
+
+            // 测试可压缩的情况
+            headers.insert("content-type", "text/plain".parse().unwrap());
+            assert!(is_compressible(&headers, "file.txt"));
+            assert!(is_compressible(&headers, "file.log"));
+
+            headers.insert("content-type", "text/html".parse().unwrap());
+            assert!(is_compressible(&headers, "file.html"));
+
+            headers.insert("content-type", "application/json".parse().unwrap());
+            assert!(is_compressible(&headers, "file.json"));
+        });
+    }
+}

ecstore/src/config/com.rs

@@ -93,17 +93,11 @@ pub async fn delete_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<()>
 }
 
 pub async fn save_config_with_opts<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<()> {
-    warn!(
-        "save_config_with_opts, bucket: {}, file: {}, data len: {}",
-        RUSTFS_META_BUCKET,
-        file,
-        data.len()
-    );
     if let Err(err) = api
         .put_object(RUSTFS_META_BUCKET, file, &mut PutObjReader::from_vec(data), opts)
         .await
     {
-        warn!("save_config_with_opts: err: {:?}, file: {}", err, file);
+        error!("save_config_with_opts: err: {:?}, file: {}", err, file);
         return Err(err);
     }
     Ok(())

ecstore/src/config/storageclass.rs

@@ -112,7 +112,13 @@ impl Config {
         }
     }
 
-    pub fn should_inline(&self, shard_size: usize, versioned: bool) -> bool {
+    pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
+        if shard_size < 0 {
+            return false;
+        }
+
+        let shard_size = shard_size as usize;
+
         let mut inline_block = DEFAULT_INLINE_BLOCK;
         if self.initialized {
             inline_block = self.inline_block;

ecstore/src/disk/local.rs

@@ -773,7 +773,7 @@ impl LocalDisk {
             Ok(res) => res,
             Err(e) => {
                 if e != DiskError::VolumeNotFound && e != Error::FileNotFound {
-                    warn!("scan list_dir {}, err {:?}", &current, &e);
+                    debug!("scan list_dir {}, err {:?}", &current, &e);
                 }
 
                 if opts.report_notfound && e == Error::FileNotFound && current == &opts.base_dir {
@@ -785,7 +785,6 @@ impl LocalDisk {
         };
 
         if entries.is_empty() {
-            warn!("scan list_dir {}, entries is empty", &current);
             return Ok(());
         }
 
@@ -801,7 +800,6 @@ impl LocalDisk {
             let entry = item.clone();
             // check limit
             if opts.limit > 0 && *objs_returned >= opts.limit {
-                warn!("scan list_dir {}, limit reached", &current);
                 return Ok(());
             }
             // check prefix
@@ -1207,7 +1205,7 @@ impl DiskAPI for LocalDisk {
             let err = self
                 .bitrot_verify(
                     &part_path,
-                    erasure.shard_file_size(part.size),
+                    erasure.shard_file_size(part.size as i64) as usize,
                     checksum_info.algorithm,
                     &checksum_info.hash,
                     erasure.shard_size(),
@@ -1248,7 +1246,7 @@ impl DiskAPI for LocalDisk {
                         resp.results[i] = CHECK_PART_FILE_NOT_FOUND;
                         continue;
                     }
-                    if (st.len() as usize) < fi.erasure.shard_file_size(part.size) {
+                    if (st.len() as i64) < fi.erasure.shard_file_size(part.size as i64) {
                         resp.results[i] = CHECK_PART_FILE_CORRUPT;
                         continue;
                     }
@@ -1400,7 +1398,7 @@ impl DiskAPI for LocalDisk {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
+    async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: i64) -> Result<FileWriter> {
         // warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path);
 
         if !origvolume.is_empty() {
@@ -1574,11 +1572,6 @@ impl DiskAPI for LocalDisk {
         let mut current = opts.base_dir.clone();
         self.scan_dir(&mut current, &opts, &mut out, &mut objs_returned).await?;
 
-        warn!(
-            "walk_dir: done, volume_dir: {:?}, base_dir: {}",
-            volume_dir.to_string_lossy(),
-            opts.base_dir
-        );
         Ok(())
     }
 
@@ -2239,7 +2232,7 @@ impl DiskAPI for LocalDisk {
                     let mut obj_deleted = false;
                     for info in obj_infos.iter() {
                         let done = ScannerMetrics::time(ScannerMetric::ApplyVersion);
-                        let sz: usize;
+                        let sz: i64;
                         (obj_deleted, sz) = item.apply_actions(info, &mut size_s).await;
                         done();
 
@@ -2260,7 +2253,7 @@ impl DiskAPI for LocalDisk {
                             size_s.versions += 1;
                         }
 
-                        size_s.total_size += sz;
+                        size_s.total_size += sz as usize;
 
                         if info.delete_marker {
                             continue;

ecstore/src/disk/mod.rs

@@ -304,7 +304,7 @@ impl DiskAPI for Disk {
     }
 
     #[tracing::instrument(skip(self))]
-    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
+    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: i64) -> 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,
@@ -491,7 +491,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     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<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>;
+    async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter>;
     // ReadFileStream
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
     async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()>;

ecstore/src/disk/remote.rs

@@ -640,7 +640,7 @@ impl DiskAPI for RemoteDisk {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter> {
+    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter> {
         info!("create_file {}/{}/{}", self.endpoint.to_string(), volume, path);
 
         let url = format!(

ecstore/src/erasure_coding/decode.rs

@@ -30,7 +30,7 @@ where
     // readers传入前应处理disk错误，确保每个reader达到可用数量的BitrotReader
     pub fn new(readers: Vec<Option<BitrotReader<R>>>, e: Erasure, offset: usize, total_length: usize) -> Self {
         let shard_size = e.shard_size();
-        let shard_file_size = e.shard_file_size(total_length);
+        let shard_file_size = e.shard_file_size(total_length as i64) as usize;
 
         let offset = (offset / e.block_size) * shard_size;
 
@@ -142,6 +142,7 @@ where
     W: tokio::io::AsyncWrite + Send + Sync + Unpin,
 {
     if get_data_block_len(en_blocks, data_blocks) < length {
+        error!("write_data_blocks get_data_block_len < length");
         return Err(io::Error::new(ErrorKind::UnexpectedEof, "Not enough data blocks to write"));
     }
 
@@ -150,6 +151,7 @@ where
 
     for block_op in &en_blocks[..data_blocks] {
         if block_op.is_none() {
+            error!("write_data_blocks block_op.is_none()");
             return Err(io::Error::new(ErrorKind::UnexpectedEof, "Missing data block"));
         }
 
@@ -164,7 +166,10 @@ where
         offset = 0;
 
         if write_left < block.len() {
-            writer.write_all(&block_slice[..write_left]).await?;
+            writer.write_all(&block_slice[..write_left]).await.map_err(|e| {
+                error!("write_data_blocks write_all err: {}", e);
+                e
+            })?;
 
             total_written += write_left;
             break;
@@ -172,7 +177,10 @@ where
 
         let n = block_slice.len();
 
-        writer.write_all(block_slice).await?;
+        writer.write_all(block_slice).await.map_err(|e| {
+            error!("write_data_blocks write_all2 err: {}", e);
+            e
+        })?;
 
         write_left -= n;
 
@@ -228,6 +236,7 @@ impl Erasure {
             };
 
             if block_length == 0 {
+                // error!("erasure decode decode block_length == 0");
                 break;
             }
 

ecstore/src/erasure_coding/erasure.rs

@@ -469,22 +469,27 @@ impl Erasure {
     }
     /// Calculate the total erasure file size for a given original size.
     // Returns the final erasure size from the original size
-    pub fn shard_file_size(&self, total_length: usize) -> usize {
+    pub fn shard_file_size(&self, total_length: i64) -> i64 {
         if total_length == 0 {
             return 0;
         }
+        if total_length < 0 {
+            return total_length;
+        }
+
+        let total_length = total_length as usize;
 
         let num_shards = total_length / self.block_size;
         let last_block_size = total_length % self.block_size;
         let last_shard_size = calc_shard_size(last_block_size, self.data_shards);
-        num_shards * self.shard_size() + last_shard_size
+        (num_shards * self.shard_size() + last_shard_size) as i64
     }
 
     /// Calculate the offset in the erasure file where reading begins.
     // Returns the offset in the erasure file where reading begins
     pub fn shard_file_offset(&self, start_offset: usize, length: usize, total_length: usize) -> usize {
         let shard_size = self.shard_size();
-        let shard_file_size = self.shard_file_size(total_length);
+        let shard_file_size = self.shard_file_size(total_length as i64) as usize;
         let end_shard = (start_offset + length) / self.block_size;
         let mut till_offset = end_shard * shard_size + shard_size;
         if till_offset > shard_file_size {

ecstore/src/heal/data_scanner.rs

@@ -526,7 +526,7 @@ impl ScannerItem {
             cumulative_size += obj_info.size;
         }
 
-        if cumulative_size >= SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE.load(Ordering::SeqCst) as usize {
+        if cumulative_size >= SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL_SIZE.load(Ordering::SeqCst) as i64 {
             //todo
         }
 
@@ -558,7 +558,7 @@ impl ScannerItem {
         Ok(object_infos)
     }
 
-    pub async fn apply_actions(&mut self, oi: &ObjectInfo, _size_s: &mut SizeSummary) -> (bool, usize) {
+    pub async fn apply_actions(&mut self, oi: &ObjectInfo, _size_s: &mut SizeSummary) -> (bool, i64) {
         let done = ScannerMetrics::time(ScannerMetric::Ilm);
         //todo: lifecycle
         info!(
@@ -641,21 +641,21 @@ impl ScannerItem {
             match tgt_status {
                 ReplicationStatusType::Pending => {
                     tgt_size_s.pending_count += 1;
-                    tgt_size_s.pending_size += oi.size;
+                    tgt_size_s.pending_size += oi.size as usize;
                     size_s.pending_count += 1;
-                    size_s.pending_size += oi.size;
+                    size_s.pending_size += oi.size as usize;
                 }
                 ReplicationStatusType::Failed => {
                     tgt_size_s.failed_count += 1;
-                    tgt_size_s.failed_size += oi.size;
+                    tgt_size_s.failed_size += oi.size as usize;
                     size_s.failed_count += 1;
-                    size_s.failed_size += oi.size;
+                    size_s.failed_size += oi.size as usize;
                 }
                 ReplicationStatusType::Completed | ReplicationStatusType::CompletedLegacy => {
                     tgt_size_s.replicated_count += 1;
-                    tgt_size_s.replicated_size += oi.size;
+                    tgt_size_s.replicated_size += oi.size as usize;
                     size_s.replicated_count += 1;
-                    size_s.replicated_size += oi.size;
+                    size_s.replicated_size += oi.size as usize;
                 }
                 _ => {}
             }
@@ -663,7 +663,7 @@ impl ScannerItem {
 
         if matches!(oi.replication_status, ReplicationStatusType::Replica) {
             size_s.replica_count += 1;
-            size_s.replica_size += oi.size;
+            size_s.replica_size += oi.size as usize;
         }
     }
 }

ecstore/src/lib.rs

@@ -4,6 +4,7 @@ pub mod bucket;
 pub mod cache_value;
 mod chunk_stream;
 pub mod cmd;
+pub mod compress;
 pub mod config;
 pub mod disk;
 pub mod disks_layout;

ecstore/src/pools.rs

@@ -24,7 +24,7 @@ use futures::future::BoxFuture;
 use http::HeaderMap;
 use rmp_serde::{Deserializer, Serializer};
 use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
-use rustfs_rio::HashReader;
+use rustfs_rio::{HashReader, WarpReader};
 use rustfs_utils::path::{SLASH_SEPARATOR, encode_dir_object, path_join};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -33,7 +33,7 @@ use std::io::{Cursor, Write};
 use std::path::PathBuf;
 use std::sync::Arc;
 use time::{Duration, OffsetDateTime};
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncReadExt, BufReader};
 use tokio::sync::broadcast::Receiver as B_Receiver;
 use tracing::{error, info, warn};
 
@@ -1254,6 +1254,7 @@ impl ECStore {
             }
 
             if let Err(err) = self
+                .clone()
                 .complete_multipart_upload(
                     &bucket,
                     &object_info.name,
@@ -1275,10 +1276,9 @@ impl ECStore {
             return Ok(());
         }
 
-        let mut data = PutObjReader::new(
-            HashReader::new(rd.stream, object_info.size as i64, object_info.size as i64, None, false)?,
-            object_info.size,
-        );
+        let reader = BufReader::new(rd.stream);
+        let hrd = HashReader::new(Box::new(WarpReader::new(reader)), object_info.size, object_info.size, None, false)?;
+        let mut data = PutObjReader::new(hrd);
 
         if let Err(err) = self
             .put_object(

ecstore/src/rebalance.rs

@@ -12,13 +12,13 @@ use crate::store_api::{CompletePart, GetObjectReader, ObjectIO, ObjectOptions, P
 use common::defer;
 use http::HeaderMap;
 use rustfs_filemeta::{FileInfo, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
-use rustfs_rio::HashReader;
+use rustfs_rio::{HashReader, WarpReader};
 use rustfs_utils::path::encode_dir_object;
 use serde::{Deserialize, Serialize};
 use std::io::Cursor;
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncReadExt, BufReader};
 use tokio::sync::broadcast::{self, Receiver as B_Receiver};
 use tokio::time::{Duration, Instant};
 use tracing::{error, info, warn};
@@ -62,7 +62,7 @@ impl RebalanceStats {
 
         self.num_versions += 1;
         let on_disk_size = if !fi.deleted {
-            fi.size as i64 * (fi.erasure.data_blocks + fi.erasure.parity_blocks) as i64 / fi.erasure.data_blocks as i64
+            fi.size * (fi.erasure.data_blocks + fi.erasure.parity_blocks) as i64 / fi.erasure.data_blocks as i64
         } else {
             0
         };
@@ -703,7 +703,7 @@ impl ECStore {
     #[allow(unused_assignments)]
     #[tracing::instrument(skip(self, set))]
     async fn rebalance_entry(
-        &self,
+        self: Arc<Self>,
         bucket: String,
         pool_index: usize,
         entry: MetaCacheEntry,
@@ -834,7 +834,7 @@ impl ECStore {
                     }
                 };
 
-                if let Err(err) = self.rebalance_object(pool_index, bucket.clone(), rd).await {
+                if let Err(err) = self.clone().rebalance_object(pool_index, bucket.clone(), rd).await {
                     if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) {
                         ignore = true;
                         warn!("rebalance_entry {} Entry {} is already deleted, skipping", &bucket, version.name);
@@ -890,7 +890,7 @@ impl ECStore {
     }
 
     #[tracing::instrument(skip(self, rd))]
-    async fn rebalance_object(&self, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> {
+    async fn rebalance_object(self: Arc<Self>, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> {
         let object_info = rd.object_info.clone();
 
         // TODO: check : use size or actual_size ?
@@ -969,6 +969,7 @@ impl ECStore {
             }
 
             if let Err(err) = self
+                .clone()
                 .complete_multipart_upload(
                     &bucket,
                     &object_info.name,
@@ -989,8 +990,9 @@ impl ECStore {
             return Ok(());
         }
 
-        let hrd = HashReader::new(rd.stream, object_info.size as i64, object_info.size as i64, None, false)?;
-        let mut data = PutObjReader::new(hrd, object_info.size);
+        let reader = BufReader::new(rd.stream);
+        let hrd = HashReader::new(Box::new(WarpReader::new(reader)), object_info.size, object_info.size, None, false)?;
+        let mut data = PutObjReader::new(hrd);
 
         if let Err(err) = self
             .put_object(

ecstore/src/set_disk.rs

@@ -55,13 +55,14 @@ use lock::{LockApi, namespace_lock::NsLockMap};
 use madmin::heal_commands::{HealDriveInfo, HealResultItem};
 use md5::{Digest as Md5Digest, Md5};
 use rand::{Rng, seq::SliceRandom};
+use rustfs_filemeta::headers::RESERVED_METADATA_PREFIX_LOWER;
 use rustfs_filemeta::{
     FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
     RawFileInfo, file_info_from_raw,
     headers::{AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
     merge_file_meta_versions,
 };
-use rustfs_rio::{EtagResolvable, HashReader};
+use rustfs_rio::{EtagResolvable, HashReader, TryGetIndex as _, WarpReader};
 use rustfs_utils::{
     HashAlgorithm,
     crypto::{base64_decode, base64_encode, hex},
@@ -860,7 +861,8 @@ impl SetDisks {
         };
 
         if let Some(err) = reduce_read_quorum_errs(errs, OBJECT_OP_IGNORED_ERRS, expected_rquorum) {
-            error!("object_quorum_from_meta: {:?}, errs={:?}", err, errs);
+            // let object = parts_metadata.first().map(|v| v.name.clone()).unwrap_or_default();
+            // error!("object_quorum_from_meta: {:?}, errs={:?}, object={:?}", err, errs, object);
             return Err(err);
         }
 
@@ -1773,7 +1775,7 @@ impl SetDisks {
         {
             Ok(v) => v,
             Err(e) => {
-                error!("Self::object_quorum_from_meta: {:?}, bucket: {}, object: {}", &e, bucket, object);
+                // error!("Self::object_quorum_from_meta: {:?}, bucket: {}, object: {}", &e, bucket, object);
                 return Err(e);
             }
         };
@@ -1817,7 +1819,7 @@ impl SetDisks {
         bucket: &str,
         object: &str,
         offset: usize,
-        length: usize,
+        length: i64,
         writer: &mut W,
         fi: FileInfo,
         files: Vec<FileInfo>,
@@ -1830,11 +1832,16 @@ impl SetDisks {
     {
         let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi);
 
-        let total_size = fi.size;
+        let total_size = fi.size as usize;
 
-        let length = { if length == 0 { total_size - offset } else { length } };
+        let length = if length < 0 {
+            fi.size as usize - offset
+        } else {
+            length as usize
+        };
 
         if offset > total_size || offset + length > total_size {
+            error!("get_object_with_fileinfo offset out of range: {}, total_size: {}", offset, total_size);
             return Err(Error::other("offset out of range"));
         }
 
@@ -1852,11 +1859,6 @@ impl SetDisks {
 
         let (last_part_index, _) = fi.to_part_offset(end_offset)?;
 
-        // debug!(
-        //     "get_object_with_fileinfo end offset:{}, last_part_index:{},part_offset:{}",
-        //     end_offset, last_part_index, 0
-        // );
-
         // let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
 
         let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
@@ -1870,7 +1872,7 @@ impl SetDisks {
             let part_number = fi.parts[i].number;
             let part_size = fi.parts[i].size;
             let mut part_length = part_size - part_offset;
-            if part_length > length - total_readed {
+            if part_length > (length - total_readed) {
                 part_length = length - total_readed
             }
 
@@ -1912,7 +1914,7 @@ impl SetDisks {
                     error!("create_bitrot_reader reduce_read_quorum_errs {:?}", &errors);
                     return Err(to_object_err(read_err.into(), vec![bucket, object]));
                 }
-
+                error!("create_bitrot_reader not enough disks to read: {:?}", &errors);
                 return Err(Error::other(format!("not enough disks to read: {:?}", errors)));
             }
 
@@ -2259,7 +2261,8 @@ impl SetDisks {
                             erasure_coding::Erasure::default()
                         };
 
-                        result.object_size = ObjectInfo::from_file_info(&lastest_meta, bucket, object, true).get_actual_size()?;
+                        result.object_size =
+                            ObjectInfo::from_file_info(&lastest_meta, bucket, object, true).get_actual_size()? as usize;
                         // Loop to find number of disks with valid data, per-drive
                         // data state and a list of outdated disks on which data needs
                         // to be healed.
@@ -2521,7 +2524,7 @@ impl SetDisks {
                                         disk.as_ref(),
                                         RUSTFS_META_TMP_BUCKET,
                                         &format!("{}/{}/part.{}", tmp_id, dst_data_dir, part.number),
-                                        erasure.shard_file_size(part.size),
+                                        erasure.shard_file_size(part.size as i64),
                                         erasure.shard_size(),
                                         HashAlgorithm::HighwayHash256,
                                     )
@@ -2603,6 +2606,7 @@ impl SetDisks {
                                         part.size,
                                         part.mod_time,
                                         part.actual_size,
+                                        part.index.clone(),
                                     );
                                     if is_inline_buffer {
                                         if let Some(writer) = writers[index].take() {
@@ -2834,7 +2838,7 @@ impl SetDisks {
             heal_item_type: HEAL_ITEM_OBJECT.to_string(),
             bucket: bucket.to_string(),
             object: object.to_string(),
-            object_size: lfi.size,
+            object_size: lfi.size as usize,
             version_id: version_id.to_string(),
             disk_count: disk_len,
             ..Default::default()
@@ -3500,7 +3504,7 @@ impl SetDisks {
                         if let (Some(started), Some(mod_time)) = (started, version.mod_time) {
                             if mod_time > started {
                                 version_not_found += 1;
-                                if send(heal_entry_skipped(version.size)).await {
+                                if send(heal_entry_skipped(version.size as usize)).await {
                                     defer.await;
                                     return;
                                 }
@@ -3544,10 +3548,10 @@ impl SetDisks {
 
                         if version_healed {
                             bg_seq.count_healed(HEAL_ITEM_OBJECT.to_string()).await;
-                            result = heal_entry_success(version.size);
+                            result = heal_entry_success(version.size as usize);
                         } else {
                             bg_seq.count_failed(HEAL_ITEM_OBJECT.to_string()).await;
-                            result = heal_entry_failure(version.size);
+                            result = heal_entry_failure(version.size as usize);
                             match version.version_id {
                                 Some(version_id) => {
                                     info!("unable to heal object {}/{}-v({})", bucket, version.name, version_id);
@@ -3863,7 +3867,7 @@ impl ObjectIO for SetDisks {
 
         let is_inline_buffer = {
             if let Some(sc) = GLOBAL_StorageClass.get() {
-                sc.should_inline(erasure.shard_file_size(data.content_length), opts.versioned)
+                sc.should_inline(erasure.shard_file_size(data.size()), opts.versioned)
             } else {
                 false
             }
@@ -3878,7 +3882,7 @@ impl ObjectIO for SetDisks {
                     Some(disk),
                     RUSTFS_META_TMP_BUCKET,
                     &tmp_object,
-                    erasure.shard_file_size(data.content_length),
+                    erasure.shard_file_size(data.size()),
                     erasure.shard_size(),
                     HashAlgorithm::HighwayHash256,
                 )
@@ -3924,7 +3928,10 @@ impl ObjectIO for SetDisks {
             return Err(Error::other(format!("not enough disks to write: {:?}", errors)));
         }
 
-        let stream = mem::replace(&mut data.stream, HashReader::new(Box::new(Cursor::new(Vec::new())), 0, 0, None, false)?);
+        let stream = mem::replace(
+            &mut data.stream,
+            HashReader::new(Box::new(WarpReader::new(Cursor::new(Vec::new()))), 0, 0, None, false)?,
+        );
 
         let (reader, w_size) = match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await {
             Ok((r, w)) => (r, w),
@@ -3939,6 +3946,16 @@ impl ObjectIO for SetDisks {
         //     error!("close_bitrot_writers err {:?}", err);
         // }
 
+        if (w_size as i64) < data.size() {
+            return Err(Error::other("put_object write size < data.size()"));
+        }
+
+        if user_defined.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER)) {
+            user_defined.insert(format!("{}compression-size", RESERVED_METADATA_PREFIX_LOWER), w_size.to_string());
+        }
+
+        let index_op = data.stream.try_get_index().map(|v| v.clone().into_vec());
+
         //TODO: userDefined
 
         let etag = data.stream.try_resolve_etag().unwrap_or_default();
@@ -3949,6 +3966,14 @@ impl ObjectIO for SetDisks {
             //  get content-type
         }
 
+        let mut actual_size = data.actual_size();
+        if actual_size < 0 {
+            let is_compressed = fi.is_compressed();
+            if !is_compressed {
+                actual_size = w_size as i64;
+            }
+        }
+
         if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS) {
             if sc == storageclass::STANDARD {
                 let _ = user_defined.remove(AMZ_STORAGE_CLASS);
@@ -3962,17 +3987,19 @@ impl ObjectIO for SetDisks {
                 if let Some(writer) = writers[i].take() {
                     fi.data = Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default());
                 }
+
+                fi.set_inline_data();
             }
 
             fi.metadata = user_defined.clone();
             fi.mod_time = Some(now);
-            fi.size = w_size;
+            fi.size = w_size as i64;
             fi.versioned = opts.versioned || opts.version_suspended;
-            fi.add_object_part(1, etag.clone(), w_size, fi.mod_time, w_size);
+            fi.add_object_part(1, etag.clone(), w_size, fi.mod_time, actual_size, index_op.clone());
 
-            fi.set_inline_data();
-
-            // debug!("put_object fi {:?}", &fi)
+            if opts.data_movement {
+                fi.set_data_moved();
+            }
         }
 
         let (online_disks, _, op_old_dir) = Self::rename_data(
@@ -4566,7 +4593,7 @@ impl StorageAPI for SetDisks {
                     Some(disk),
                     RUSTFS_META_TMP_BUCKET,
                     &tmp_part_path,
-                    erasure.shard_file_size(data.content_length),
+                    erasure.shard_file_size(data.size()),
                     erasure.shard_size(),
                     HashAlgorithm::HighwayHash256,
                 )
@@ -4605,16 +4632,33 @@ impl StorageAPI for SetDisks {
             return Err(Error::other(format!("not enough disks to write: {:?}", errors)));
         }
 
-        let stream = mem::replace(&mut data.stream, HashReader::new(Box::new(Cursor::new(Vec::new())), 0, 0, None, false)?);
+        let stream = mem::replace(
+            &mut data.stream,
+            HashReader::new(Box::new(WarpReader::new(Cursor::new(Vec::new()))), 0, 0, None, false)?,
+        );
 
         let (reader, w_size) = Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?; // TODO: 出错，删除临时目录
 
         let _ = mem::replace(&mut data.stream, reader);
 
+        if (w_size as i64) < data.size() {
+            return Err(Error::other("put_object_part write size < data.size()"));
+        }
+
+        let index_op = data.stream.try_get_index().map(|v| v.clone().into_vec());
+
         let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
 
         if let Some(ref tag) = opts.preserve_etag {
-            etag = tag.clone(); // TODO: 需要验证 etag 是否一致
+            etag = tag.clone();
+        }
+
+        let mut actual_size = data.actual_size();
+        if actual_size < 0 {
+            let is_compressed = fi.is_compressed();
+            if !is_compressed {
+                actual_size = w_size as i64;
+            }
         }
 
         let part_info = ObjectPartInfo {
@@ -4622,7 +4666,8 @@ impl StorageAPI for SetDisks {
             number: part_id,
             size: w_size,
             mod_time: Some(OffsetDateTime::now_utc()),
-            actual_size: data.content_length,
+            actual_size,
+            index: index_op,
             ..Default::default()
         };
 
@@ -4649,6 +4694,7 @@ impl StorageAPI for SetDisks {
             part_num: part_id,
             last_mod: Some(OffsetDateTime::now_utc()),
             size: w_size,
+            actual_size,
         };
 
         // error!("put_object_part ret {:?}", &ret);
@@ -4932,7 +4978,7 @@ impl StorageAPI for SetDisks {
     // complete_multipart_upload 完成
     #[tracing::instrument(skip(self))]
     async fn complete_multipart_upload(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         object: &str,
         upload_id: &str,
@@ -4974,12 +5020,15 @@ impl StorageAPI for SetDisks {
         for (i, res) in part_files_resp.iter().enumerate() {
             let part_id = uploaded_parts[i].part_num;
             if !res.error.is_empty() || !res.exists {
-                // error!("complete_multipart_upload part_id err {:?}", res);
+                error!("complete_multipart_upload part_id err {:?}, exists={}", res, res.exists);
                 return Err(Error::InvalidPart(part_id, bucket.to_owned(), object.to_owned()));
             }
 
-            let part_fi = FileInfo::unmarshal(&res.data).map_err(|_e| {
-                // error!("complete_multipart_upload FileInfo::unmarshal err {:?}", e);
+            let part_fi = FileInfo::unmarshal(&res.data).map_err(|e| {
+                error!(
+                    "complete_multipart_upload FileInfo::unmarshal err {:?}, part_id={}, bucket={}, object={}",
+                    e, part_id, bucket, object
+                );
                 Error::InvalidPart(part_id, bucket.to_owned(), object.to_owned())
             })?;
             let part = &part_fi.parts[0];
@@ -4989,11 +5038,18 @@ impl StorageAPI for SetDisks {
             // debug!("complete part {} object info {:?}", part_num, &part);
 
             if part_id != part_num {
-                // error!("complete_multipart_upload part_id err part_id != part_num {} != {}", part_id, part_num);
+                error!("complete_multipart_upload part_id err part_id != part_num {} != {}", part_id, part_num);
                 return Err(Error::InvalidPart(part_id, bucket.to_owned(), object.to_owned()));
             }
 
-            fi.add_object_part(part.number, part.etag.clone(), part.size, part.mod_time, part.actual_size);
+            fi.add_object_part(
+                part.number,
+                part.etag.clone(),
+                part.size,
+                part.mod_time,
+                part.actual_size,
+                part.index.clone(),
+            );
         }
 
         let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files_metas, &fi);
@@ -5003,24 +5059,35 @@ impl StorageAPI for SetDisks {
         fi.parts = Vec::with_capacity(uploaded_parts.len());
 
         let mut object_size: usize = 0;
-        let mut object_actual_size: usize = 0;
+        let mut object_actual_size: i64 = 0;
 
         for (i, p) in uploaded_parts.iter().enumerate() {
             let has_part = curr_fi.parts.iter().find(|v| v.number == p.part_num);
             if has_part.is_none() {
-                // error!("complete_multipart_upload has_part.is_none() {:?}", has_part);
+                error!(
+                    "complete_multipart_upload has_part.is_none() {:?}, part_id={}, bucket={}, object={}",
+                    has_part, p.part_num, bucket, object
+                );
                 return Err(Error::InvalidPart(p.part_num, "".to_owned(), p.etag.clone().unwrap_or_default()));
             }
 
             let ext_part = &curr_fi.parts[i];
 
             if p.etag != Some(ext_part.etag.clone()) {
+                error!(
+                    "complete_multipart_upload etag err {:?}, part_id={}, bucket={}, object={}",
+                    p.etag, p.part_num, bucket, object
+                );
                 return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
             }
 
             // TODO: crypto
 
-            if (i < uploaded_parts.len() - 1) && !is_min_allowed_part_size(ext_part.size) {
+            if (i < uploaded_parts.len() - 1) && !is_min_allowed_part_size(ext_part.actual_size) {
+                error!(
+                    "complete_multipart_upload is_min_allowed_part_size err {:?}, part_id={}, bucket={}, object={}",
+                    ext_part.actual_size, p.part_num, bucket, object
+                );
                 return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
             }
 
@@ -5033,11 +5100,12 @@ impl StorageAPI for SetDisks {
                 size: ext_part.size,
                 mod_time: ext_part.mod_time,
                 actual_size: ext_part.actual_size,
+                index: ext_part.index.clone(),
                 ..Default::default()
             });
         }
 
-        fi.size = object_size;
+        fi.size = object_size as i64;
         fi.mod_time = opts.mod_time;
         if fi.mod_time.is_none() {
             fi.mod_time = Some(OffsetDateTime::now_utc());
@@ -5054,6 +5122,18 @@ impl StorageAPI for SetDisks {
 
         fi.metadata.insert("etag".to_owned(), etag);
 
+        fi.metadata
+            .insert(format!("{}actual-size", RESERVED_METADATA_PREFIX_LOWER), object_actual_size.to_string());
+
+        if fi.is_compressed() {
+            fi.metadata
+                .insert(format!("{}compression-size", RESERVED_METADATA_PREFIX_LOWER), object_size.to_string());
+        }
+
+        if opts.data_movement {
+            fi.set_data_moved();
+        }
+
         // TODO: object_actual_size
         let _ = object_actual_size;
 
@@ -5125,17 +5205,6 @@ impl StorageAPI for SetDisks {
         )
         .await?;
 
-        for (i, op_disk) in online_disks.iter().enumerate() {
-            if let Some(disk) = op_disk {
-                if disk.is_online().await {
-                    fi = parts_metadatas[i].clone();
-                    break;
-                }
-            }
-        }
-
-        fi.is_latest = true;
-
         // debug!("complete fileinfo {:?}", &fi);
 
         // TODO: reduce_common_data_dir
@@ -5157,7 +5226,22 @@ impl StorageAPI for SetDisks {
                 .await;
         }
 
-        let _ = self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await;
+        let upload_id_path = upload_id_path.clone();
+        let store = self.clone();
+        let _cleanup_handle = tokio::spawn(async move {
+            let _ = store.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await;
+        });
+
+        for (i, op_disk) in online_disks.iter().enumerate() {
+            if let Some(disk) = op_disk {
+                if disk.is_online().await {
+                    fi = parts_metadatas[i].clone();
+                    break;
+                }
+            }
+        }
+
+        fi.is_latest = true;
 
         Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
@@ -5517,7 +5601,7 @@ async fn disks_with_all_parts(
                 let verify_err = bitrot_verify(
                     Box::new(Cursor::new(data.clone())),
                     data_len,
-                    meta.erasure.shard_file_size(meta.size),
+                    meta.erasure.shard_file_size(meta.size) as usize,
                     checksum_info.algorithm,
                     checksum_info.hash,
                     meta.erasure.shard_size(),
@@ -5729,8 +5813,8 @@ pub async fn stat_all_dirs(disks: &[Option<DiskStore>], bucket: &str, prefix: &s
 }
 
 const GLOBAL_MIN_PART_SIZE: ByteSize = ByteSize::mib(5);
-fn is_min_allowed_part_size(size: usize) -> bool {
-    size as u64 >= GLOBAL_MIN_PART_SIZE.as_u64()
+fn is_min_allowed_part_size(size: i64) -> bool {
+    size >= GLOBAL_MIN_PART_SIZE.as_u64() as i64
 }
 
 fn get_complete_multipart_md5(parts: &[CompletePart]) -> String {

ecstore/src/sets.rs

@@ -627,7 +627,7 @@ impl StorageAPI for Sets {
 
     #[tracing::instrument(skip(self))]
     async fn complete_multipart_upload(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         object: &str,
         upload_id: &str,

ecstore/src/store.rs

@@ -1233,7 +1233,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.size()).await?;
 
         if opts.data_movement && idx == opts.src_pool_idx {
             return Err(StorageError::DataMovementOverwriteErr(
@@ -1508,9 +1508,7 @@ impl StorageAPI for ECStore {
 
         // TODO: nslock
 
-        let pool_idx = self
-            .get_pool_idx_no_lock(src_bucket, &src_object, src_info.size as i64)
-            .await?;
+        let pool_idx = self.get_pool_idx_no_lock(src_bucket, &src_object, src_info.size).await?;
 
         if cp_src_dst_same {
             if let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id) {
@@ -1995,7 +1993,7 @@ impl StorageAPI for ECStore {
 
     #[tracing::instrument(skip(self))]
     async fn complete_multipart_upload(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         object: &str,
         upload_id: &str,
@@ -2006,6 +2004,7 @@ impl StorageAPI for ECStore {
 
         if self.single_pool() {
             return self.pools[0]
+                .clone()
                 .complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
                 .await;
         }
@@ -2015,6 +2014,7 @@ impl StorageAPI for ECStore {
                 continue;
             }
 
+            let pool = pool.clone();
             let err = match pool
                 .complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts)
                 .await

ecstore/src/store_api.rs

@@ -7,24 +7,24 @@ use crate::store_utils::clean_metadata;
 use crate::{disk::DiskStore, heal::heal_commands::HealOpts};
 use http::{HeaderMap, HeaderValue};
 use madmin::heal_commands::HealResultItem;
+use rustfs_filemeta::headers::RESERVED_METADATA_PREFIX_LOWER;
 use rustfs_filemeta::{FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, headers::AMZ_OBJECT_TAGGING};
-use rustfs_rio::{HashReader, Reader};
+use rustfs_rio::{DecompressReader, HashReader, LimitReader, WarpReader};
+use rustfs_utils::CompressionAlgorithm;
 use rustfs_utils::path::decode_dir_object;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::io::Cursor;
+use std::str::FromStr as _;
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncRead, AsyncReadExt};
+use tracing::warn;
 use uuid::Uuid;
 
 pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
 pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M
-pub const RESERVED_METADATA_PREFIX: &str = "X-Rustfs-Internal-";
-pub const RESERVED_METADATA_PREFIX_LOWER: &str = "x-rustfs-internal-";
-pub const RUSTFS_HEALING: &str = "X-Rustfs-Internal-healing";
-pub const RUSTFS_DATA_MOVE: &str = "X-Rustfs-Internal-data-mov";
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct MakeBucketOptions {
@@ -53,46 +53,50 @@ pub struct DeleteBucketOptions {
 
 pub struct PutObjReader {
     pub stream: HashReader,
-    pub content_length: usize,
 }
 
 impl Debug for PutObjReader {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("PutObjReader")
-            .field("content_length", &self.content_length)
-            .finish()
+        f.debug_struct("PutObjReader").finish()
     }
 }
 
 impl PutObjReader {
-    pub fn new(stream: HashReader, content_length: usize) -> Self {
-        PutObjReader { stream, content_length }
+    pub fn new(stream: HashReader) -> Self {
+        PutObjReader { stream }
     }
 
     pub fn from_vec(data: Vec<u8>) -> Self {
-        let content_length = data.len();
+        let content_length = data.len() as i64;
         PutObjReader {
-            stream: HashReader::new(Box::new(Cursor::new(data)), content_length as i64, content_length as i64, None, false)
+            stream: HashReader::new(Box::new(WarpReader::new(Cursor::new(data))), content_length, content_length, None, false)
                 .unwrap(),
-            content_length,
         }
     }
+
+    pub fn size(&self) -> i64 {
+        self.stream.size()
+    }
+
+    pub fn actual_size(&self) -> i64 {
+        self.stream.actual_size()
+    }
 }
 
 pub struct GetObjectReader {
-    pub stream: Box<dyn Reader>,
+    pub stream: Box<dyn AsyncRead + Unpin + Send + Sync>,
     pub object_info: ObjectInfo,
 }
 
 impl GetObjectReader {
     #[tracing::instrument(level = "debug", skip(reader))]
     pub fn new(
-        reader: Box<dyn Reader>,
+        reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
         rs: Option<HTTPRangeSpec>,
         oi: &ObjectInfo,
         opts: &ObjectOptions,
         _h: &HeaderMap<HeaderValue>,
-    ) -> Result<(Self, usize, usize)> {
+    ) -> Result<(Self, usize, i64)> {
         let mut rs = rs;
 
         if let Some(part_number) = opts.part_number {
@@ -101,6 +105,47 @@ impl GetObjectReader {
             }
         }
 
+        // TODO:Encrypted
+
+        let (algo, is_compressed) = oi.is_compressed_ok()?;
+
+        // TODO: check TRANSITION
+
+        if is_compressed {
+            let actual_size = oi.get_actual_size()?;
+            let (off, length) = (0, oi.size);
+            let (_dec_off, dec_length) = (0, actual_size);
+            if let Some(_rs) = rs {
+                // TODO: range spec is not supported for compressed object
+                return Err(Error::other("The requested range is not satisfiable"));
+                // let (off, length) = rs.get_offset_length(actual_size)?;
+            }
+
+            let dec_reader = DecompressReader::new(reader, algo);
+
+            let actual_size = if actual_size > 0 {
+                actual_size as usize
+            } else {
+                return Err(Error::other(format!("invalid decompressed size {}", actual_size)));
+            };
+
+            warn!("actual_size: {}", actual_size);
+            let dec_reader = LimitReader::new(dec_reader, actual_size);
+
+            let mut oi = oi.clone();
+            oi.size = dec_length;
+
+            warn!("oi.size: {}, off: {}, length: {}", oi.size, off, length);
+            return Ok((
+                GetObjectReader {
+                    stream: Box::new(dec_reader),
+                    object_info: oi,
+                },
+                off,
+                length,
+            ));
+        }
+
         if let Some(rs) = rs {
             let (off, length) = rs.get_offset_length(oi.size)?;
 
@@ -142,8 +187,8 @@ impl GetObjectReader {
 #[derive(Debug)]
 pub struct HTTPRangeSpec {
     pub is_suffix_length: bool,
-    pub start: usize,
-    pub end: Option<usize>,
+    pub start: i64,
+    pub end: i64,
 }
 
 impl HTTPRangeSpec {
@@ -152,29 +197,38 @@ impl HTTPRangeSpec {
             return None;
         }
 
-        let mut start = 0;
-        let mut end = -1;
+        let mut start = 0i64;
+        let mut end = -1i64;
         for i in 0..oi.parts.len().min(part_number) {
             start = end + 1;
-            end = start + oi.parts[i].size as i64 - 1
+            end = start + (oi.parts[i].size as i64) - 1
         }
 
         Some(HTTPRangeSpec {
             is_suffix_length: false,
-            start: start as usize,
-            end: { if end < 0 { None } else { Some(end as usize) } },
+            start,
+            end,
         })
     }
 
-    pub fn get_offset_length(&self, res_size: usize) -> Result<(usize, usize)> {
+    pub fn get_offset_length(&self, res_size: i64) -> Result<(usize, i64)> {
         let len = self.get_length(res_size)?;
+
         let mut start = self.start;
         if self.is_suffix_length {
-            start = res_size - self.start
+            start = res_size + self.start;
+
+            if start < 0 {
+                start = 0;
+            }
         }
-        Ok((start, len))
+        Ok((start as usize, len))
     }
-    pub fn get_length(&self, res_size: usize) -> Result<usize> {
+    pub fn get_length(&self, res_size: i64) -> Result<i64> {
+        if res_size < 0 {
+            return Err(Error::other("The requested range is not satisfiable"));
+        }
+
         if self.is_suffix_length {
             let specified_len = self.start; // 假设 h.start 是一个 i64 类型
             let mut range_length = specified_len;
@@ -190,8 +244,8 @@ impl HTTPRangeSpec {
             return Err(Error::other("The requested range is not satisfiable"));
         }
 
-        if let Some(end) = self.end {
-            let mut end = end;
+        if self.end > -1 {
+            let mut end = self.end;
             if res_size <= end {
                 end = res_size - 1;
             }
@@ -200,7 +254,7 @@ impl HTTPRangeSpec {
             return Ok(range_length);
         }
 
-        if self.end.is_none() {
+        if self.end == -1 {
             let range_length = res_size - self.start;
             return Ok(range_length);
         }
@@ -276,6 +330,7 @@ pub struct PartInfo {
     pub last_mod: Option<OffsetDateTime>,
     pub size: usize,
     pub etag: Option<String>,
+    pub actual_size: i64,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -298,9 +353,9 @@ pub struct ObjectInfo {
     pub bucket: String,
     pub name: String,
     pub mod_time: Option<OffsetDateTime>,
-    pub size: usize,
+    pub size: i64,
     // Actual size is the real size of the object uploaded by client.
-    pub actual_size: Option<usize>,
+    pub actual_size: i64,
     pub is_dir: bool,
     pub user_defined: Option<HashMap<String, String>>,
     pub parity_blocks: usize,
@@ -364,27 +419,41 @@ impl Clone for ObjectInfo {
 impl ObjectInfo {
     pub fn is_compressed(&self) -> bool {
         if let Some(meta) = &self.user_defined {
-            meta.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
+            meta.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER))
         } else {
             false
         }
     }
 
+    pub fn is_compressed_ok(&self) -> Result<(CompressionAlgorithm, bool)> {
+        let scheme = self
+            .user_defined
+            .as_ref()
+            .and_then(|meta| meta.get(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER)).cloned());
+
+        if let Some(scheme) = scheme {
+            let algorithm = CompressionAlgorithm::from_str(&scheme)?;
+            Ok((algorithm, true))
+        } else {
+            Ok((CompressionAlgorithm::None, false))
+        }
+    }
+
     pub fn is_multipart(&self) -> bool {
         self.etag.as_ref().is_some_and(|v| v.len() != 32)
     }
 
-    pub fn get_actual_size(&self) -> std::io::Result<usize> {
-        if let Some(actual_size) = self.actual_size {
-            return Ok(actual_size);
+    pub fn get_actual_size(&self) -> std::io::Result<i64> {
+        if self.actual_size > 0 {
+            return Ok(self.actual_size);
         }
 
         if self.is_compressed() {
             if let Some(meta) = &self.user_defined {
-                if let Some(size_str) = meta.get(&format!("{}actual-size", RESERVED_METADATA_PREFIX)) {
+                if let Some(size_str) = meta.get(&format!("{}actual-size", RESERVED_METADATA_PREFIX_LOWER)) {
                     if !size_str.is_empty() {
                         // Todo: deal with error
-                        let size = size_str.parse::<usize>().map_err(|e| std::io::Error::other(e.to_string()))?;
+                        let size = size_str.parse::<i64>().map_err(|e| std::io::Error::other(e.to_string()))?;
                         return Ok(size);
                     }
                 }
@@ -395,8 +464,9 @@ impl ObjectInfo {
                 actual_size += part.actual_size;
             });
             if actual_size == 0 && actual_size != self.size {
-                return Err(std::io::Error::other("invalid decompressed size"));
+                return Err(std::io::Error::other(format!("invalid decompressed size {} {}", actual_size, self.size)));
             }
+
             return Ok(actual_size);
         }
 
@@ -803,7 +873,7 @@ pub trait StorageAPI: ObjectIO {
     // ListObjectParts
     async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()>;
     async fn complete_multipart_upload(
-        &self,
+        self: Arc<Self>,
         bucket: &str,
         object: &str,
         upload_id: &str,