From 2a3d127a867995412235a99440c87f659a7d0c17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Tue, 14 Apr 2026 16:35:03 +0800 Subject: [PATCH] revert: restore PutObjReader and remove PUT zero-copy chunk path (#2351 phase 2) (#2532) --- crates/ecstore/src/bucket/migration.rs | 6 +- crates/ecstore/src/config/com.rs | 16 +- crates/ecstore/src/data_movement.rs | 10 +- crates/ecstore/src/set_disk.rs | 60 +++--- crates/ecstore/src/set_disk/write.rs | 132 ------------- crates/ecstore/src/sets.rs | 18 +- crates/ecstore/src/store.rs | 17 +- crates/ecstore/src/store/multipart.rs | 2 +- crates/ecstore/src/store/object.rs | 2 +- crates/ecstore/src/store_api/readers.rs | 173 ++---------------- crates/ecstore/src/store_api/traits.rs | 10 +- crates/ecstore/src/store_api/types.rs | 2 +- crates/ecstore/src/tier/tier.rs | 8 +- crates/heal/src/heal/storage.rs | 2 +- crates/heal/tests/heal_integration_test.rs | 4 +- crates/protocols/src/swift/object.rs | 6 +- .../tests/lifecycle_integration_test.rs | 14 +- rustfs/src/app/capacity_dirty_scope_test.rs | 8 +- .../src/app/lifecycle_transition_api_test.rs | 8 +- rustfs/src/app/multipart_usecase.rs | 10 +- rustfs/src/app/object_usecase.rs | 4 +- .../app/object_usecase/put_object_extract.rs | 2 +- .../src/app/object_usecase/put_object_flow.rs | 4 +- 23 files changed, 114 insertions(+), 404 deletions(-) diff --git a/crates/ecstore/src/bucket/migration.rs b/crates/ecstore/src/bucket/migration.rs index 15d01a45b..1a91c7967 100644 --- a/crates/ecstore/src/bucket/migration.rs +++ b/crates/ecstore/src/bucket/migration.rs @@ -17,7 +17,7 @@ use crate::bucket::metadata::BUCKET_METADATA_FILE; use crate::bucket::replication::{decode_resync_file, encode_resync_file}; use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; -use crate::store_api::{BucketOptions, ChunkNativePutData, ObjectOptions, StorageAPI}; +use crate::store_api::{BucketOptions, ObjectOptions, PutObjReader, StorageAPI}; use http::HeaderMap; use rustfs_policy::auth::UserIdentity; use rustfs_policy::policy::PolicyDoc; @@ -263,7 +263,7 @@ async fn migrate_one_if_missing( } }; - let mut put_data = ChunkNativePutData::from_vec(data); + let mut put_data = PutObjReader::from_vec(data); if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await { warn!("write {label}: {e}"); } else { @@ -341,7 +341,7 @@ pub async fn try_migrate_iam_config(store: Arc) { continue; } }; - let mut put_data = ChunkNativePutData::from_vec(data); + let mut put_data = PutObjReader::from_vec(data); if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await { warn!("write IAM config {path}: {e}"); } else { diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index f2e48c2fe..2f506a81c 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -16,7 +16,7 @@ use crate::config::{Config, GLOBAL_STORAGE_CLASS, KVS, audit, notify, oidc, stor use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::global::is_first_cluster_node_local; -use crate::store_api::{ChunkNativePutData, ObjectInfo, ObjectOptions, StorageAPI}; +use crate::store_api::{ObjectInfo, ObjectOptions, PutObjReader, StorageAPI}; use http::HeaderMap; use rustfs_config::audit::{AUDIT_MQTT_KEYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_KEYS, AUDIT_WEBHOOK_SUB_SYS}; use rustfs_config::notify::{NOTIFY_MQTT_KEYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_WEBHOOK_KEYS, NOTIFY_WEBHOOK_SUB_SYS}; @@ -128,7 +128,7 @@ pub async fn delete_config(api: Arc, file: &str) -> Result<()> } pub async fn save_config_with_opts(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result<()> { - let mut put_data = ChunkNativePutData::from_vec(data); + let mut put_data = PutObjReader::from_vec(data); if let Err(err) = api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await { error!("save_config_with_opts: err: {:?}, file: {}", err, file); return Err(err); @@ -1076,10 +1076,10 @@ mod tests { use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type}; use crate::set_disk::SetDisks; use crate::store_api::{ - BucketInfo, BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, DeleteBucketOptions, DeletedObject, - GetObjectReader, HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectVersionsInfo, ListObjectsV2Info, - ListOperations, MakeBucketOptions, MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, - ObjectOperations, ObjectOptions, ObjectToDelete, PartInfo, StorageAPI, WalkOptions, + BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, + HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectVersionsInfo, ListObjectsV2Info, ListOperations, + MakeBucketOptions, MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOperations, + ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI, WalkOptions, }; use http::HeaderMap; use rustfs_config::audit::{AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS}; @@ -1302,7 +1302,7 @@ mod tests { &self, _bucket: &str, _object: &str, - _data: &mut ChunkNativePutData, + _data: &mut PutObjReader, _opts: &ObjectOptions, ) -> Result { panic!("unused in test") @@ -1489,7 +1489,7 @@ mod tests { _object: &str, _upload_id: &str, _part_id: usize, - _data: &mut ChunkNativePutData, + _data: &mut PutObjReader, _opts: &ObjectOptions, ) -> Result { panic!("unused in test") diff --git a/crates/ecstore/src/data_movement.rs b/crates/ecstore/src/data_movement.rs index 7d2ef7d5f..0e1fcbf0b 100644 --- a/crates/ecstore/src/data_movement.rs +++ b/crates/ecstore/src/data_movement.rs @@ -14,9 +14,7 @@ use crate::error::{Error, Result}; use crate::store::ECStore; -use crate::store_api::{ - ChunkNativePutData, CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOptions, -}; +use crate::store_api::{CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader}; use bytes::Bytes; use rustfs_rio::{BlockReadable, BoxReadBlockFuture, EtagResolvable, HashReader, HashReaderDetector, Index, TryGetIndex}; use std::io::Cursor; @@ -71,7 +69,7 @@ pub fn decode_part_index(index: Option<&Bytes>) -> Option { } } -pub fn put_data_from_chunk(chunk: Vec, size: i64, actual_size: i64, index: Option) -> Result { +pub fn put_data_from_chunk(chunk: Vec, size: i64, actual_size: i64, index: Option) -> Result { use sha2::{Digest, Sha256}; let sha256hex = if !chunk.is_empty() { @@ -82,7 +80,7 @@ pub fn put_data_from_chunk(chunk: Vec, size: i64, actual_size: i64, index: O let reader = IndexedDataMovementReader::new(Cursor::new(chunk), index); let hash_reader = HashReader::from_reader(reader, size, actual_size, None, sha256hex, false)?; - Ok(ChunkNativePutData::new(hash_reader)) + Ok(PutObjReader::new(hash_reader)) } pub fn new_multipart_abort_flag() -> Arc { @@ -262,7 +260,7 @@ pub(crate) async fn migrate_object( .and_then(|part| decode_part_index(part.index.as_ref())); let reader = IndexedDataMovementReader::new(BufReader::new(rd.stream), index); let hrd = HashReader::from_reader(reader, object_info.size, actual_size, object_info.etag.clone(), None, false)?; - let mut data = ChunkNativePutData::new(hrd); + let mut data = PutObjReader::new(hrd); if let Err(err) = store .put_object( diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index d73144aba..4548d3118 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -52,9 +52,9 @@ use crate::{ event_notification::{EventArgs, send_event}, global::{GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_deployment_id, is_dist_erasure}, store_api::{ - BucketInfo, BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, DeleteBucketOptions, DeletedObject, - GetObjectReader, HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectsV2Info, ListOperations, MakeBucketOptions, - MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOperations, PartInfo, StorageAPI, + BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, + HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectsV2Info, ListOperations, MakeBucketOptions, MultipartInfo, + MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOperations, PartInfo, PutObjReader, StorageAPI, }, store_init::load_format_erasure, }; @@ -790,13 +790,7 @@ impl ObjectIO for SetDisks { } #[tracing::instrument(skip(self, data,))] - async fn put_object( - &self, - bucket: &str, - object: &str, - data: &mut ChunkNativePutData, - opts: &ObjectOptions, - ) -> Result { + async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { let disks = self.get_disks_internal().await; let mut object_lock_guard = None; @@ -945,8 +939,14 @@ impl ObjectIO for SetDisks { let object_size = data.size(); let encode_write_start = Instant::now(); - let w_size = match Self::write_chunk_native_put_data(data, Arc::new(erasure), &mut writers, write_quorum).await { - Ok(written) => written, + + let stream = mem::replace( + &mut data.stream, + HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, + ); + + let (reader, w_size) = match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { + Ok((r, w)) => (r, w), Err(e) => { log_put_storage_phase( bucket, @@ -961,12 +961,14 @@ impl ObjectIO for SetDisks { return Err(e.into()); } }; // TODO: delete temporary directory on error + + let _ = mem::replace(&mut data.stream, reader); log_put_storage_phase( bucket, object, "encode_write", encode_write_start.elapsed(), - data.size(), + object_size, is_inline_buffer, write_quorum, ); @@ -984,11 +986,11 @@ impl ObjectIO for SetDisks { insert_str(&mut user_defined, SUFFIX_COMPRESSION_SIZE, w_size.to_string()); } - let index_op = data.index_bytes(); + let index_op = data.stream.try_get_index().map(|v| v.clone().into_vec()); //TODO: userDefined - let etag = data.resolve_etag().unwrap_or_default(); + let etag = data.stream.try_resolve_etag().unwrap_or_default(); user_defined.insert("etag".to_owned(), etag.clone()); @@ -1005,9 +1007,9 @@ impl ObjectIO for SetDisks { } if fi.checksum.is_none() - && let Some(content_hash) = data.content_hash_bytes()? + && let Some(content_hash) = data.as_hash_reader().content_hash() { - fi.checksum = Some(content_hash); + fi.checksum = Some(content_hash.to_bytes(&[])); } if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS) @@ -2266,7 +2268,7 @@ impl ObjectOperations for SetDisks { let gr = gr.unwrap(); let reader = BufReader::new(gr.stream); let hash_reader = HashReader::from_stream(reader, gr.object_info.size, gr.object_info.size, None, None, false)?; - let mut p_reader = ChunkNativePutData::new(hash_reader); + let mut p_reader = PutObjReader::new(hash_reader); return match self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await { Ok(restored_info) => { send_event(EventArgs { @@ -2334,7 +2336,7 @@ impl ObjectOperations for SetDisks { }; let reader = BufReader::new(gr.stream); let hash_reader = HashReader::from_stream(reader, part_info.actual_size, part_info.actual_size, None, None, false)?; - let mut p_reader = ChunkNativePutData::new(hash_reader); + let mut p_reader = PutObjReader::new(hash_reader); let p_info = self_ .clone() .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &ObjectOptions::default()) @@ -2558,7 +2560,7 @@ impl MultipartOperations for SetDisks { object: &str, upload_id: &str, part_id: usize, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); @@ -2570,8 +2572,9 @@ impl MultipartOperations for SetDisks { if let Some(checksum) = fi.metadata.get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM) && !checksum.is_empty() && data + .as_hash_reader() .content_crc_type() - .is_none_or(|v: rustfs_rio::ChecksumType| v.to_string() != *checksum) + .is_none_or(|v| v.to_string() != *checksum) { return Err(Error::other(format!("checksum mismatch: {checksum}"))); } @@ -2636,7 +2639,14 @@ impl MultipartOperations for SetDisks { return Err(Error::other(format!("not enough disks to write: {errors:?}"))); } - let w_size = Self::write_chunk_native_put_data(data, Arc::new(erasure), &mut writers, write_quorum).await?; // TODO: delete temporary directory on error + let stream = mem::replace( + &mut data.stream, + HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, + ); + + let (reader, w_size) = Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?; // TODO: delete temporary directory on error + + let _ = mem::replace(&mut data.stream, reader); if (w_size as i64) < data.size() { warn!("put_object_part write size < data.size(), w_size={}, data.size={}", w_size, data.size()); @@ -2647,9 +2657,9 @@ impl MultipartOperations for SetDisks { ))); } - let index_op = data.index_bytes(); + let index_op = data.stream.try_get_index().map(|v| v.clone().into_vec()); - let mut etag = data.resolve_etag().unwrap_or_default(); + let mut etag = data.stream.try_resolve_etag().unwrap_or_default(); if let Some(ref tag) = opts.preserve_etag { etag = tag.clone(); @@ -2663,7 +2673,7 @@ impl MultipartOperations for SetDisks { } } - let checksums = data.content_crc(); + let checksums = data.as_hash_reader().content_crc(); let part_info = ObjectPartInfo { etag: etag.clone(), diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index 7c0122ee4..d937ede3e 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -13,87 +13,12 @@ // limitations under the License. use super::*; -use crate::store_api::ChunkNativePutData; impl SetDisks { - fn all_inline_bitrot_writers(writers: &[Option]) -> bool { - writers.iter().all(|writer| { - writer - .as_ref() - .is_some_and(crate::erasure_coding::BitrotWriterWrapper::is_inline_buffer) - }) - } - - async fn write_chunk_native_put_data_inline( - data: &mut ChunkNativePutData, - erasure: Arc, - writers: &mut [Option], - write_quorum: usize, - ) -> std::io::Result { - let stream = data.take_stream()?; - let mut assembler = erasure_coding::encode::BlockAssembler::new(stream, erasure.block_size); - let encoder = erasure_coding::encode::ErasureChunkEncoder::new(erasure).await; - let mut writer_group = erasure_coding::encode::MultiWriter::new(writers, write_quorum); - - loop { - let block = match assembler.next_block().await { - Ok(Some(block)) => block, - Ok(None) => break, - Err(err) => { - data.restore_stream(assembler.into_inner()); - return Err(err); - } - }; - - let encoded = match encoder.encode_block(&block).await { - Ok(encoded) => encoded, - Err(err) => { - data.restore_stream(assembler.into_inner()); - return Err(err); - } - }; - - if let Err(err) = writer_group.write_inline(&encoded) { - encoder.release(encoded).await; - data.restore_stream(assembler.into_inner()); - return Err(err); - } - - encoder.release(encoded).await; - } - - let total_bytes = assembler.total_bytes(); - let stream = assembler.into_inner(); - if let Err(err) = writer_group.shutdown_inline() { - data.restore_stream(stream); - return Err(err); - } - data.restore_stream(stream); - Ok(total_bytes) - } - pub(super) fn default_read_quorum(&self) -> usize { self.set_drive_count - self.default_parity_count } - pub(super) async fn write_chunk_native_put_data( - data: &mut ChunkNativePutData, - erasure: Arc, - writers: &mut [Option], - write_quorum: usize, - ) -> std::io::Result { - if Self::all_inline_bitrot_writers(writers) { - return Self::write_chunk_native_put_data_inline(data, erasure, writers, write_quorum).await; - } - - let stream = data.take_stream()?; - let (stream, written) = erasure_coding::encode::ErasureWritePipeline::new(erasure, write_quorum) - .run(stream, writers) - .await?; - data.restore_stream(stream); - Ok(written) - } - pub(super) fn default_write_quorum(&self) -> usize { let mut data_count = self.set_drive_count - self.default_parity_count; if data_count == self.default_parity_count { @@ -702,60 +627,3 @@ impl SetDisks { None } } - -#[cfg(test)] -mod tests { - use super::*; - use crate::erasure_coding::{BitrotWriterWrapper, CustomWriter}; - use crate::store_api::ChunkNativePutData; - - #[tokio::test] - async fn write_chunk_native_put_data_restores_reader_state_after_encoding() { - let payload = b"chunk-native-put-payload".repeat(8); - let erasure = Arc::new(erasure_coding::Erasure::new(2, 1, 8)); - let mut reader = ChunkNativePutData::from_vec(payload.clone()); - let mut writers: Vec> = (0..erasure.total_shard_count()) - .map(|_| { - Some(BitrotWriterWrapper::new( - CustomWriter::new_inline_buffer(), - erasure.shard_size(), - HashAlgorithm::HighwayHash256S, - )) - }) - .collect(); - - let written = SetDisks::write_chunk_native_put_data(&mut reader, erasure.clone(), &mut writers, 2) - .await - .unwrap(); - - assert_eq!(written, payload.len()); - assert_eq!(reader.size(), payload.len() as i64); - assert_eq!(reader.actual_size(), payload.len() as i64); - assert!( - reader.resolve_etag().is_some(), - "restored reader should preserve computed etag state after chunk-native encode" - ); - - let inline_lengths: Vec = writers - .into_iter() - .map(|writer| writer.expect("writer").into_inline_data().expect("inline data").len()) - .collect(); - assert!(inline_lengths.iter().all(|len| *len > 0)); - } - - #[tokio::test] - async fn write_chunk_native_put_data_detects_inline_writer_set() { - let erasure = Arc::new(erasure_coding::Erasure::new(2, 1, 8)); - let writers: Vec> = (0..erasure.total_shard_count()) - .map(|_| { - Some(BitrotWriterWrapper::new( - CustomWriter::new_inline_buffer(), - erasure.shard_size(), - HashAlgorithm::HighwayHash256S, - )) - }) - .collect(); - - assert!(SetDisks::all_inline_bitrot_writers(&writers)); - } -} diff --git a/crates/ecstore/src/sets.rs b/crates/ecstore/src/sets.rs index 974441142..48502a4b6 100644 --- a/crates/ecstore/src/sets.rs +++ b/crates/ecstore/src/sets.rs @@ -28,10 +28,10 @@ use crate::{ global::{GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_lock_clients, is_dist_erasure}, set_disk::SetDisks, store_api::{ - BucketInfo, BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, DeleteBucketOptions, DeletedObject, - GetObjectReader, HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectVersionsInfo, ListObjectsV2Info, - ListOperations, MakeBucketOptions, MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, - ObjectOperations, ObjectOptions, ObjectToDelete, PartInfo, StorageAPI, + BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, + HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectVersionsInfo, ListObjectsV2Info, ListOperations, + MakeBucketOptions, MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOperations, + ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI, }, store_init::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file}, }; @@ -375,13 +375,7 @@ impl ObjectIO for Sets { .await } #[tracing::instrument(level = "debug", skip(self, data))] - async fn put_object( - &self, - bucket: &str, - object: &str, - data: &mut ChunkNativePutData, - opts: &ObjectOptions, - ) -> Result { + async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).put_object(bucket, object, data, opts).await } } @@ -694,7 +688,7 @@ impl MultipartOperations for Sets { object: &str, upload_id: &str, part_id: usize, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { self.get_disks_by_key(object) diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index baba75643..07e16baa4 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -59,10 +59,9 @@ use crate::{ rpc::S3PeerSys, sets::Sets, store_api::{ - BucketInfo, BucketOperations, BucketOptions, ChunkNativePutData, CompletePart, DeleteBucketOptions, DeletedObject, - GetObjectReader, HTTPRangeSpec, HealOperations, ListObjectsV2Info, ListOperations, MakeBucketOptions, - MultipartOperations, MultipartUploadResult, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, PartInfo, - StorageAPI, + BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, + HTTPRangeSpec, HealOperations, ListObjectsV2Info, ListOperations, MakeBucketOptions, MultipartOperations, + MultipartUploadResult, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI, }, store_init, }; @@ -260,13 +259,7 @@ impl ObjectIO for ECStore { self.handle_get_object_reader(bucket, object, range, h, opts).await } #[instrument(level = "debug", skip(self, data))] - async fn put_object( - &self, - bucket: &str, - object: &str, - data: &mut ChunkNativePutData, - opts: &ObjectOptions, - ) -> Result { + async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { enqueue_transition_after_write(self.handle_put_object(bucket, object, data, opts).await, LcEventSrc::S3PutObject).await } } @@ -502,7 +495,7 @@ impl MultipartOperations for ECStore { object: &str, upload_id: &str, part_id: usize, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { self.handle_put_object_part(bucket, object, upload_id, part_id, data, opts) diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 12ef5712f..dfeb1c143 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -176,7 +176,7 @@ impl ECStore { object: &str, upload_id: &str, part_id: usize, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { check_put_object_part_args(bucket, object, upload_id)?; diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 1686e63cf..1d6ba282f 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -217,7 +217,7 @@ impl ECStore { &self, bucket: &str, object: &str, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { check_put_object_args(bucket, object)?; diff --git a/crates/ecstore/src/store_api/readers.rs b/crates/ecstore/src/store_api/readers.rs index e843216a8..01f3bbb65 100644 --- a/crates/ecstore/src/store_api/readers.rs +++ b/crates/ecstore/src/store_api/readers.rs @@ -1,101 +1,7 @@ use super::*; -use rustfs_rio::TryGetIndex; - -pub struct ChunkNativePutData { - stream: Option, - size: i64, - actual_size: i64, -} - -impl Debug for ChunkNativePutData { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("ChunkNativePutData").finish() - } -} - -impl ChunkNativePutData { - pub fn new(stream: HashReader) -> Self { - let size = stream.size(); - let actual_size = stream.actual_size(); - Self { - stream: Some(stream), - size, - actual_size, - } - } - - pub fn from_vec(data: Vec) -> Self { - use sha2::{Digest, Sha256}; - - let content_length = data.len() as i64; - let sha256hex = if content_length > 0 { - Some(hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower)) - } else { - None - }; - Self::new(HashReader::from_stream(Cursor::new(data), content_length, content_length, None, sha256hex, false).unwrap()) - } - - pub fn take_stream(&mut self) -> std::io::Result { - self.stream - .take() - .ok_or_else(|| std::io::Error::other("ChunkNativePutData stream already taken")) - } - - pub fn restore_stream(&mut self, stream: HashReader) { - self.size = stream.size(); - self.actual_size = stream.actual_size(); - self.stream = Some(stream); - } - - pub fn as_hash_reader(&self) -> Option<&HashReader> { - self.stream.as_ref() - } - - pub fn as_hash_reader_mut(&mut self) -> Option<&mut HashReader> { - self.stream.as_mut() - } - - pub fn index_bytes(&self) -> Option { - self.as_hash_reader() - .and_then(|reader| reader.try_get_index().map(|index| index.clone().into_vec())) - } - - pub fn resolve_etag(&mut self) -> Option { - self.as_hash_reader_mut() - .and_then(rustfs_rio::EtagResolvable::try_resolve_etag) - } - - pub fn content_hash_bytes(&mut self) -> std::io::Result> { - let Some(reader) = self.as_hash_reader_mut() else { - return Ok(None); - }; - - Ok(reader - .finalize_content_hash()? - .as_ref() - .map(|checksum| checksum.to_bytes(&[]))) - } - - pub fn content_crc_type(&self) -> Option { - self.as_hash_reader().and_then(HashReader::content_crc_type) - } - - pub fn content_crc(&self) -> HashMap { - self.as_hash_reader().map_or_else(HashMap::new, HashReader::content_crc) - } - - pub fn size(&self) -> i64 { - self.size - } - - pub fn actual_size(&self) -> i64 { - self.actual_size - } -} pub struct PutObjReader { - data: ChunkNativePutData, + pub stream: HashReader, } impl Debug for PutObjReader { @@ -106,81 +12,32 @@ impl Debug for PutObjReader { impl PutObjReader { pub fn new(stream: HashReader) -> Self { - Self { - data: ChunkNativePutData::new(stream), - } + PutObjReader { stream } } - pub fn chunk_native_data(&self) -> &ChunkNativePutData { - &self.data - } - - pub fn chunk_native_data_mut(&mut self) -> &mut ChunkNativePutData { - &mut self.data - } - - pub fn take_stream(&mut self) -> std::io::Result { - self.data.take_stream() - } - - pub fn restore_stream(&mut self, stream: HashReader) { - self.data.restore_stream(stream); - } - - pub fn as_hash_reader(&self) -> Option<&HashReader> { - self.data.as_hash_reader() - } - - pub fn as_hash_reader_mut(&mut self) -> Option<&mut HashReader> { - self.data.as_hash_reader_mut() - } - - pub fn index_bytes(&self) -> Option { - self.data.index_bytes() - } - - pub fn resolve_etag(&mut self) -> Option { - self.data.resolve_etag() - } - - pub fn content_hash_bytes(&mut self) -> std::io::Result> { - self.data.content_hash_bytes() - } - - pub fn content_crc_type(&self) -> Option { - self.data.content_crc_type() - } - - pub fn content_crc(&self) -> HashMap { - self.data.content_crc() + pub fn as_hash_reader(&self) -> &HashReader { + &self.stream } pub fn from_vec(data: Vec) -> Self { - Self { - data: ChunkNativePutData::from_vec(data), + use sha2::{Digest, Sha256}; + let content_length = data.len() as i64; + let sha256hex = if content_length > 0 { + Some(hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower)) + } else { + None + }; + PutObjReader { + stream: HashReader::from_stream(Cursor::new(data), content_length, content_length, None, sha256hex, false).unwrap(), } } pub fn size(&self) -> i64 { - self.data.size() + self.stream.size() } pub fn actual_size(&self) -> i64 { - self.data.actual_size() - } -} - -impl std::ops::Deref for PutObjReader { - type Target = ChunkNativePutData; - - fn deref(&self) -> &Self::Target { - &self.data - } -} - -impl std::ops::DerefMut for PutObjReader { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.data + self.stream.actual_size() } } diff --git a/crates/ecstore/src/store_api/traits.rs b/crates/ecstore/src/store_api/traits.rs index a351976e8..facabeae3 100644 --- a/crates/ecstore/src/store_api/traits.rs +++ b/crates/ecstore/src/store_api/traits.rs @@ -11,13 +11,7 @@ pub trait ObjectIO: Send + Sync + Debug + 'static { opts: &ObjectOptions, ) -> Result; - async fn put_object( - &self, - bucket: &str, - object: &str, - data: &mut ChunkNativePutData, - opts: &ObjectOptions, - ) -> Result; + async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result; } /// Bucket-level storage operations. @@ -132,7 +126,7 @@ pub trait MultipartOperations: Send + Sync + Debug { object: &str, upload_id: &str, part_id: usize, - data: &mut ChunkNativePutData, + data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result; async fn get_multipart_info( diff --git a/crates/ecstore/src/store_api/types.rs b/crates/ecstore/src/store_api/types.rs index 51bd69d61..b569dca70 100644 --- a/crates/ecstore/src/store_api/types.rs +++ b/crates/ecstore/src/store_api/types.rs @@ -285,7 +285,7 @@ pub struct ObjectInfo { pub expires: Option, pub num_versions: usize, pub successor_mod_time: Option, - pub put_object_reader: Option, + pub put_object_reader: Option, pub etag: Option, pub inlined: bool, pub metadata_only: bool, diff --git a/crates/ecstore/src/tier/tier.rs b/crates/ecstore/src/tier/tier.rs index 97b546f38..e5dfd8b42 100644 --- a/crates/ecstore/src/tier/tier.rs +++ b/crates/ecstore/src/tier/tier.rs @@ -51,7 +51,7 @@ use crate::{ disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}, global::is_first_cluster_node_local, store::ECStore, - store_api::{ChunkNativePutData, ObjectIO as _, ObjectOptions}, + store_api::{ObjectIO as _, ObjectOptions, PutObjReader}, }; use rustfs_rio::HashReader; use rustfs_utils::path::{SLASH_SEPARATOR, path_join}; @@ -1046,7 +1046,7 @@ impl TierConfigMgr { opts: &ObjectOptions, ) -> std::result::Result<(), std::io::Error> { debug!("save tier config:{}", file); - let mut put_data = ChunkNativePutData::from_vec(data.to_vec()); + let mut put_data = PutObjReader::from_vec(data.to_vec()); let _ = api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await?; Ok(()) } @@ -1103,7 +1103,7 @@ async fn load_tier_config(api: Arc) -> std::result::Result { let cfg = TierConfigMgr::unmarshal(&data)?; let normalized = encode_external_tiering_config_blob(&cfg)?; - let mut put_data = ChunkNativePutData::from_vec(normalized.to_vec()); + let mut put_data = PutObjReader::from_vec(normalized.to_vec()); let _ = api .put_object( RUSTFS_META_BUCKET, @@ -1158,7 +1158,7 @@ async fn read_tier_config_from_bucket( } async fn write_tier_config_to_rustfs(api: Arc, path: &str, data: Bytes) -> io::Result<()> { - let mut put_data = ChunkNativePutData::from_vec(data.to_vec()); + let mut put_data = PutObjReader::from_vec(data.to_vec()); api.put_object( RUSTFS_META_BUCKET, path, diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index e1699c192..33f6210bc 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -208,7 +208,7 @@ impl HealStorageAPI for ECStoreHealStorage { async fn put_object_data(&self, bucket: &str, object: &str, data: &[u8]) -> Result<()> { debug!("Putting object data: {}/{} ({} bytes)", bucket, object, data.len()); - let mut reader = rustfs_ecstore::store_api::ChunkNativePutData::from_vec(data.to_vec()); + let mut reader = rustfs_ecstore::store_api::PutObjReader::from_vec(data.to_vec()); match (*self.ecstore) .put_object(bucket, object, &mut reader, &Default::default()) .await diff --git a/crates/heal/tests/heal_integration_test.rs b/crates/heal/tests/heal_integration_test.rs index 765221408..e7ba323ef 100644 --- a/crates/heal/tests/heal_integration_test.rs +++ b/crates/heal/tests/heal_integration_test.rs @@ -18,7 +18,7 @@ use rustfs_ecstore::{ disk::endpoint::Endpoint, endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, store::ECStore, - store_api::{BucketOperations, ChunkNativePutData, ObjectIO, ObjectOperations, ObjectOptions}, + store_api::{BucketOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader}, }; use rustfs_heal::heal::{ manager::{HealConfig, HealManager}, @@ -162,7 +162,7 @@ async fn create_test_bucket(ecstore: &Arc, bucket_name: &str) { /// Test helper: Upload test object async fn upload_test_object(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) { - let mut reader = ChunkNativePutData::from_vec(data.to_vec()); + let mut reader = PutObjReader::from_vec(data.to_vec()); let object_info = (**ecstore) .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await diff --git a/crates/protocols/src/swift/object.rs b/crates/protocols/src/swift/object.rs index 763a8f202..c7715a602 100644 --- a/crates/protocols/src/swift/object.rs +++ b/crates/protocols/src/swift/object.rs @@ -55,7 +55,7 @@ use super::{SwiftError, SwiftResult}; use axum::http::HeaderMap; use rustfs_credentials::Credentials; use rustfs_ecstore::new_object_layer_fn; -use rustfs_ecstore::store_api::{BucketOperations, BucketOptions, ChunkNativePutData, ObjectIO, ObjectOperations, ObjectOptions}; +use rustfs_ecstore::store_api::{BucketOperations, BucketOptions, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader}; use rustfs_rio::HashReader; use std::collections::HashMap; use tracing::debug; @@ -382,7 +382,7 @@ where .map_err(|e| sanitize_storage_error("Hash reader creation", e))?; // 15. Hand the hash reader to the chunk-native PUT data wrapper - let mut put_reader = ChunkNativePutData::new(hash_reader); + let mut put_reader = PutObjReader::new(hash_reader); // 16. Upload object to storage let obj_info = store @@ -465,7 +465,7 @@ where .map_err(|e| sanitize_storage_error("Hash reader creation", e))?; // Hand the hash reader to the chunk-native PUT data wrapper - let mut put_reader = ChunkNativePutData::new(hash_reader); + let mut put_reader = PutObjReader::new(hash_reader); // Upload object to storage let obj_info = store diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index ca0d475ed..b8e8304f7 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -24,7 +24,7 @@ use rustfs_ecstore::{ pools::path2_bucket_object_with_base_path, store::ECStore, store_api::{ - BucketOperations, ChunkNativePutData, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, + BucketOperations, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, }, tier::{ tier_config::{TierConfig, TierMinIO, TierType}, @@ -235,7 +235,7 @@ async fn create_test_lock_bucket(ecstore: &Arc, bucket_name: &str) { /// Test helper: Upload test object async fn upload_test_object(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) { - let mut reader = ChunkNativePutData::from_vec(data.to_vec()); + let mut reader = PutObjReader::from_vec(data.to_vec()); let object_info = (**ecstore) .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await @@ -781,7 +781,7 @@ mod serial_tests { .await .expect("Failed to set lifecycle configuration"); - let mut reader = ChunkNativePutData::from_vec(put_payload.to_vec()); + let mut reader = PutObjReader::from_vec(put_payload.to_vec()); let mut metadata = HashMap::new(); metadata.insert("content-type".to_string(), "text/plain".to_string()); ecstore @@ -838,7 +838,7 @@ mod serial_tests { .expect("Failed to create multipart upload"); let part_data = b"multipart immediate transition"; - let mut reader = ChunkNativePutData::from_vec(part_data.to_vec()); + let mut reader = PutObjReader::from_vec(part_data.to_vec()); let part = ecstore .put_object_part( multipart_bucket.as_str(), @@ -903,7 +903,7 @@ mod serial_tests { .get_object_info(src_bucket.as_str(), src_object, &ObjectOptions::default()) .await .expect("Failed to load source object info"); - src_info.put_object_reader = Some(ChunkNativePutData::from_vec(payload.to_vec())); + src_info.put_object_reader = Some(PutObjReader::from_vec(payload.to_vec())); ecstore .copy_object( @@ -969,7 +969,7 @@ mod serial_tests { .await .expect("Failed to create multipart upload"); - let mut part1_reader = ChunkNativePutData::from_vec(part1); + let mut part1_reader = PutObjReader::from_vec(part1); let uploaded_part1 = ecstore .put_object_part( bucket_name.as_str(), @@ -982,7 +982,7 @@ mod serial_tests { .await .expect("Failed to upload first multipart part"); - let mut part2_reader = ChunkNativePutData::from_vec(part2); + let mut part2_reader = PutObjReader::from_vec(part2); let uploaded_part2 = ecstore .put_object_part( bucket_name.as_str(), diff --git a/rustfs/src/app/capacity_dirty_scope_test.rs b/rustfs/src/app/capacity_dirty_scope_test.rs index ad751340e..229252315 100644 --- a/rustfs/src/app/capacity_dirty_scope_test.rs +++ b/rustfs/src/app/capacity_dirty_scope_test.rs @@ -18,9 +18,7 @@ use rustfs_ecstore::{ disk::endpoint::Endpoint, endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, store::ECStore, - store_api::{ - BucketOperations, BucketOptions, ChunkNativePutData, HealOperations, MakeBucketOptions, ObjectIO, ObjectOptions, - }, + store_api::{BucketOperations, BucketOptions, HealOperations, MakeBucketOptions, ObjectIO, ObjectOptions, PutObjReader}, }; use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager}; use serial_test::serial; @@ -137,7 +135,7 @@ async fn data_movement_put_object_marks_dirty_disks_for_capacity_manager() { let _ = manager.get_dirty_disks().await; let payload = b"data-movement-dirty-scope".to_vec(); - let mut reader = ChunkNativePutData::from_vec(payload); + let mut reader = PutObjReader::from_vec(payload); let opts = ObjectOptions { data_movement: true, src_pool_idx: 0, @@ -179,7 +177,7 @@ async fn heal_object_marks_missing_shard_disk_dirty_for_capacity_manager() { let payload_len = 3 * 1024 * 1024 + 137; let payload: Vec = (0..payload_len).map(|idx| (idx % 251) as u8).collect(); - let mut reader = ChunkNativePutData::from_vec(payload); + let mut reader = PutObjReader::from_vec(payload); let object_name = "test/heal.bin"; let put_info = ecstore .put_object(&bucket_name, object_name, &mut reader, &ObjectOptions::default()) diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 03193697c..ee6524ef9 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -27,8 +27,8 @@ use rustfs_ecstore::{ global::GLOBAL_TierConfigMgr, store::ECStore, store_api::{ - BucketOperations, BucketOptions, ChunkNativePutData, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, - ObjectOptions, + BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, + PutObjReader, }, tier::{ tier_config::{TierConfig, TierType}, @@ -150,7 +150,7 @@ async fn upload_test_object( object: &str, data: &[u8], ) -> rustfs_ecstore::store_api::ObjectInfo { - let mut reader = ChunkNativePutData::from_vec(data.to_vec()); + let mut reader = PutObjReader::from_vec(data.to_vec()); (**ecstore) .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await @@ -448,7 +448,7 @@ async fn complete_multipart_upload_transitions_immediately_via_usecase() { .await .expect("Failed to create multipart upload"); - let mut reader = ChunkNativePutData::from_vec(payload.to_vec()); + let mut reader = PutObjReader::from_vec(payload.to_vec()); let uploaded_part = ecstore .put_object_part(bucket.as_str(), object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) .await diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 42b0bf840..788d1141f 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -45,9 +45,7 @@ use rustfs_ecstore::compress::is_compressible; use rustfs_ecstore::error::{StorageError, is_err_object_not_found, is_err_version_not_found}; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::set_disk::is_valid_storage_class; -use rustfs_ecstore::store_api::{ - ChunkNativePutData, CompletePart, HTTPRangeSpec, MultipartUploadResult, ObjectIO, ObjectOptions, -}; +use rustfs_ecstore::store_api::{CompletePart, HTTPRangeSpec, MultipartUploadResult, ObjectIO, ObjectOptions, PutObjReader}; use rustfs_ecstore::store_api::{MultipartOperations, ObjectOperations}; use rustfs_filemeta::{ReplicationStatusType, ReplicationType}; use rustfs_object_io::put::PutObjectChecksums; @@ -772,7 +770,7 @@ impl DefaultMultipartUsecase { None => (None, None), }; - let mut reader = ChunkNativePutData::new(reader); + let mut reader = PutObjReader::new(reader); let info = store .put_object_part(&bucket, &key, &upload_id, part_id, &mut reader, &opts) @@ -813,7 +811,7 @@ impl DefaultMultipartUsecase { _ => (), } } - checksums.merge_from_map(&reader.content_crc()); + checksums.merge_from_map(&reader.as_hash_reader().content_crc()); let output = UploadPartOutput { server_side_encryption: requested_sse, @@ -1095,7 +1093,7 @@ impl DefaultMultipartUsecase { None => (None, None), }; - let mut reader = ChunkNativePutData::new(reader); + let mut reader = PutObjReader::new(reader); let dst_opts = ObjectOptions { user_defined: mp_info.user_defined.clone(), diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index f81043b07..4ccc40dae 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -73,7 +73,7 @@ use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found, is_err_object use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::set_disk::is_valid_storage_class; use rustfs_ecstore::store_api::{ - ChunkNativePutData, HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, + HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, PutObjReader, }; use rustfs_filemeta::{ REPLICATE_INCOMING_DELETE, ReplicationStatusType, ReplicationType, RestoreStatusOps, VersionPurgeStatusType, @@ -1236,7 +1236,7 @@ impl DefaultObjectUsecase { src_info.user_defined.extend(material.metadata); } - src_info.put_object_reader = Some(ChunkNativePutData::new(reader)); + src_info.put_object_reader = Some(PutObjReader::new(reader)); // check quota diff --git a/rustfs/src/app/object_usecase/put_object_extract.rs b/rustfs/src/app/object_usecase/put_object_extract.rs index 050162691..46caf7a75 100644 --- a/rustfs/src/app/object_usecase/put_object_extract.rs +++ b/rustfs/src/app/object_usecase/put_object_extract.rs @@ -295,7 +295,7 @@ impl DefaultObjectUsecase { opts.user_defined.extend(metadata); let capacity_scope_token = Uuid::new_v4(); opts.capacity_scope_token = Some(capacity_scope_token); - let mut reader = ChunkNativePutData::new(hrd); + let mut reader = PutObjReader::new(hrd); let obj_info = match store.put_object(&bucket, &fpath, &mut reader, &opts).await { Ok(info) => info, diff --git a/rustfs/src/app/object_usecase/put_object_flow.rs b/rustfs/src/app/object_usecase/put_object_flow.rs index 500c4ddc2..be4025700 100644 --- a/rustfs/src/app/object_usecase/put_object_flow.rs +++ b/rustfs/src/app/object_usecase/put_object_flow.rs @@ -539,7 +539,7 @@ impl DefaultObjectUsecase { opts.user_defined.extend(encryption_metadata); } - let mut reader = ChunkNativePutData::new(reader); + let mut reader = PutObjReader::new(reader); let mt2 = metadata.clone(); opts.user_defined.extend(metadata); @@ -611,7 +611,7 @@ impl DefaultObjectUsecase { &request_context.trailing_headers, &mut checksums, ); - checksums.merge_from_map(&reader.content_crc()); + checksums.merge_from_map(&reader.as_hash_reader().content_crc()); if let Some(checksum_bytes) = resolved_checksum_bytes(&checksums) && obj_info .checksum