From 31df11beb7a760204217a3cf332b88dd7dd038af Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 3 Jul 2026 08:06:17 +0800 Subject: [PATCH] refactor(replication): isolate multipart and target adapters (#4211) * refactor(replication): isolate multipart part planning * refactor(replication): isolate target head adapters --- .../ecstore/src/bucket/replication/README.md | 2 +- .../replication/replication_resyncer.rs | 230 +++--------------- .../replication_target_boundary.rs | 136 ++++++++++- crates/replication/src/lib.rs | 5 + crates/replication/src/multipart.rs | 222 +++++++++++++++++ scripts/check_architecture_migration_rules.sh | 6 +- 6 files changed, 402 insertions(+), 199 deletions(-) create mode 100644 crates/replication/src/multipart.rs diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index c7b6563bb..41966b7ee 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -34,7 +34,7 @@ paths. | `ReplicationConfigStore` | Replication config persistence and config-derived labels used by target options. | Config read/save helpers and storage class labels are exposed through the contract type in `replication_config_store.rs`. | | `ReplicationFileMeta` | Replication status, decisions, MRF entries, resync decisions, and target reset helpers. | `rustfs_filemeta` replication contracts are concentrated in `replication_filemeta_boundary.rs`; `FileInfo` remains in the storage boundary for storage trait bindings and walk options. | | `ReplicationErrorBoundary` | ECStore error/result contracts and replication-specific error classifiers. | `crate::error` imports are concentrated in `replication_error_boundary.rs`. | -| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, target config types, and target operation option types. | Bucket target sys access, `BucketTargets`, and target operation types are exposed through the contract type in `replication_target_boundary.rs`. | +| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, target config types, target operation option types, and target HeadObject comparison adapters. | Bucket target sys access, `BucketTargets`, target operation types, and HeadObject-to-replication DTO adapters are exposed through the contract type in `replication_target_boundary.rs`. | | `ReplicationRuntime` | Worker pool, queue sizing, stats, bucket monitor, local node identity, cancellation, and admission state. | Direct runtime source/global access and shared replication pool/stat state; ECStore object store and bucket monitor implementation types stay behind local storage/bandwidth boundaries. | | `ReplicationBandwidthLimiter` | Target reader wrapping for replication bandwidth accounting and throttling. | Direct bucket bandwidth reader imports from resyncer paths. | | `ReplicationEventSink` | Notification and audit events for skipped, failed, pending, and completed replication operations. | Event notification service calls and local event host selection are concentrated in `replication_event_sink.rs`. | diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 0503f4ce1..b79903f89 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -34,9 +34,10 @@ use super::replication_storage_boundary::{ ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions, }; use super::replication_target_boundary::{ - PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_complete_multipart_options, - replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options, - replication_put_object_header_size, replication_put_object_options, + PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head, + replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, + replication_force_delete_remove_options, replication_put_object_header_size, replication_put_object_options, + replication_target_head_is_newer_null_version, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -52,14 +53,12 @@ use http_body::Frame; use http_body_util::StreamBody; #[cfg(test)] use rmp_serde; -#[cfg(test)] -use rustfs_replication::content_matches_by_etag; use rustfs_replication::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationSourceObject, - ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, heal_uses_delete_replication_path, - is_retryable_delete_replication_head_error, is_version_delete_replication, is_version_id_mismatch, - replication_action_for_target, replication_etags_match, resync_state_accepts_update, should_count_head_proxy_failure, - should_retry_delete_marker_purge, target_is_newer_than_source_null_version, + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationMultipartPartInput, ResyncOpts, + TargetReplicationResyncStatus, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, + is_version_delete_replication, is_version_id_mismatch, replication_etags_match, replication_multipart_complete_actual_size, + replication_multipart_part_plan, resync_state_accepts_update, should_count_head_proxy_failure, + should_retry_delete_marker_purge, }; use rustfs_s3_types::EventName; use rustfs_utils::http::{ @@ -183,38 +182,6 @@ async fn head_object_fallback( } } -fn head_object_last_modified(oi: &HeadObjectOutput) -> Option { - oi.last_modified - .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) -} - -fn replication_source_object(oi: &ObjectInfo) -> ReplicationSourceObject<'_> { - ReplicationSourceObject { - mod_time: oi.mod_time, - version_id: oi.version_id.map(|version_id| version_id.to_string()), - etag: oi.etag.as_deref(), - actual_size: oi.get_actual_size().unwrap_or_default(), - delete_marker: oi.delete_marker, - content_type: oi.content_type.as_deref(), - content_encoding: oi.content_encoding.as_deref(), - user_tags: oi.user_tags.as_str(), - user_defined: oi.user_defined.as_ref(), - } -} - -fn replication_target_object(oi: &HeadObjectOutput) -> ReplicationTargetObject<'_> { - ReplicationTargetObject { - last_modified: head_object_last_modified(oi), - version_id: oi.version_id.as_deref(), - etag: oi.e_tag.as_deref(), - content_length: oi.content_length.unwrap_or_default(), - delete_marker: oi.delete_marker.unwrap_or_default(), - content_type: oi.content_type.as_deref(), - metadata: oi.metadata.as_ref(), - tag_count: oi.tag_count.unwrap_or_default(), - } -} - fn map_replication_error(err: rustfs_replication::Error) -> Error { match err { rustfs_replication::Error::CorruptedFormat => Error::CorruptedFormat, @@ -2267,11 +2234,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await { Ok(oi) => { - replication_action = replication_action_for_target( - &replication_source_object(&object_info), - &replication_target_object(&oi), - self.op_type, - ); + replication_action = replication_action_for_target_head(&object_info, &oi, self.op_type); if replication_action == ReplicationAction::None { rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; @@ -2588,18 +2551,11 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await { Ok(oi) => { - replication_action = replication_action_for_target( - &replication_source_object(&object_info), - &replication_target_object(&oi), - self.op_type, - ); + replication_action = replication_action_for_target_head(&object_info, &oi, self.op_type); rinfo.replication_status = ReplicationStatusType::Completed; if replication_action == ReplicationAction::None { if self.op_type == ReplicationType::ExistingObject - && target_is_newer_than_source_null_version( - &replication_source_object(&object_info), - &replication_target_object(&oi), - ) + && replication_target_head_is_newer_null_version(&object_info, &oi) { warn!( event = EVENT_RESYNC_RUNTIME_SKIPPED, @@ -2885,29 +2841,6 @@ fn async_read_to_bytestream(reader: impl AsyncRead + Send + Sync + Unpin + 'stat ByteStream::new(SdkBody::from_body_1_x(body)) } -fn part_range_spec_from_actual_size(offset: i64, part_size: i64) -> std::io::Result<(HTTPRangeSpec, i64)> { - if offset < 0 { - return Err(std::io::Error::other("invalid part offset")); - } - if part_size <= 0 { - return Err(std::io::Error::other(format!("invalid part size {part_size}"))); - } - let end = offset - .checked_add(part_size - 1) - .ok_or_else(|| std::io::Error::other("part range overflow"))?; - let next_offset = end - .checked_add(1) - .ok_or_else(|| std::io::Error::other("part offset overflow"))?; - Ok(( - HTTPRangeSpec { - is_suffix_length: false, - start: offset, - end, - }, - next_offset, - )) -} - struct MultipartReplicationContext<'a, S: ReplicationObjectIO> { storage: Arc, cli: Arc, @@ -2956,11 +2889,18 @@ async fn replicate_object_with_multipart(ctx: MultipartR let mut header_size = replication_put_object_header_size(&put_opts); let mut offset: i64 = 0; for part_info in object_info.parts.iter() { - let part_number = i32::try_from(part_info.number) - .map_err(|_| std::io::Error::other(format!("part number {} overflows i32", part_info.number)))?; - let part_size = part_info.actual_size; - let (range_spec, next_offset) = part_range_spec_from_actual_size(offset, part_size)?; - offset = next_offset; + let part_plan = replication_multipart_part_plan(ReplicationMultipartPartInput { + offset, + part_number: part_info.number, + part_size: part_info.actual_size, + }) + .map_err(|err| std::io::Error::other(err.to_string()))?; + let range_spec = HTTPRangeSpec { + is_suffix_length: false, + start: part_plan.range.start, + end: part_plan.range.end, + }; + offset = part_plan.next_offset; let part_reader = storage .get_object_reader(src_bucket, object, Some(range_spec), HeaderMap::new(), obj_opts) @@ -2976,8 +2916,8 @@ async fn replicate_object_with_multipart(ctx: MultipartR dst_bucket, object, &upload_id, - part_number, - part_size, + part_plan.part_number, + part_plan.part_size, byte_stream, &PutObjectPartOptions { ..Default::default() }, ) @@ -2986,11 +2926,15 @@ async fn replicate_object_with_multipart(ctx: MultipartR let etag = object_part.e_tag.unwrap_or_default(); - uploaded_parts.push(CompletedPart::builder().part_number(part_number).e_tag(etag).build()); + uploaded_parts.push( + CompletedPart::builder() + .part_number(part_plan.part_number) + .e_tag(etag) + .build(), + ); } - let actual_size = - rustfs_utils::http::get_str(&object_info.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE).unwrap_or_default(); + let actual_size = replication_multipart_complete_actual_size(&object_info.user_defined); cli.complete_multipart_upload( dst_bucket, @@ -3008,25 +2952,10 @@ async fn replicate_object_with_multipart(ctx: MultipartR #[cfg(test)] mod tests { use super::*; - use aws_smithy_types::DateTime; use std::collections::HashMap; - use time::{Duration, OffsetDateTime}; + use time::OffsetDateTime; use uuid::Uuid; - #[test] - fn test_part_range_spec_from_actual_size() { - let (rs, next) = part_range_spec_from_actual_size(0, 10).unwrap(); - assert_eq!(rs.start, 0); - assert_eq!(rs.end, 9); - assert_eq!(next, 10); - } - - #[test] - fn test_part_range_spec_rejects_non_positive() { - assert!(part_range_spec_from_actual_size(0, 0).is_err()); - assert!(part_range_spec_from_actual_size(0, -1).is_err()); - } - #[test] fn test_unmarshal_resync_payload() { let start = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid ts"); @@ -3198,46 +3127,6 @@ mod tests { assert_eq!(target.resync_before_date, None); } - #[test] - fn test_get_replication_action_existing_object_source_newer_null_version_requires_replication() { - let source = ObjectInfo { - mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(20)), - version_id: None, - ..Default::default() - }; - let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(10)).build(); - - assert_eq!( - replication_action_for_target( - &replication_source_object(&source), - &replication_target_object(&target), - ReplicationType::ExistingObject, - ), - ReplicationAction::All, - "a newer source null version must not be skipped during existing-object replication" - ); - } - - #[test] - fn test_get_replication_action_existing_object_target_newer_null_version_skips() { - let source = ObjectInfo { - mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), - version_id: None, - ..Default::default() - }; - let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(20)).build(); - - assert_eq!( - replication_action_for_target( - &replication_source_object(&source), - &replication_target_object(&target), - ReplicationType::ExistingObject, - ), - ReplicationAction::None, - "a newer target null-version object should not be overwritten by existing-object replication" - ); - } - #[test] fn test_replicate_object_info_to_object_info_preserves_delete_marker_flag() { let live = ReplicateObjectInfo { @@ -3421,57 +3310,6 @@ mod tests { ); } - #[test] - fn test_content_matches_compares_etag_only() { - let src = ObjectInfo { - etag: Some("\"abc123\"".to_string()), - ..Default::default() - }; - - let tgt_match = HeadObjectOutput::builder().e_tag("\"abc123\"").build(); - assert!( - content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_match)), - "identical ETags must match" - ); - - let tgt_unquoted_match = HeadObjectOutput::builder().e_tag("abc123").build(); - assert!( - content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_unquoted_match)), - "quoted and unquoted ETags with identical values must match" - ); - - // version_id on the target is intentionally ignored - let tgt_different_version = HeadObjectOutput::builder() - .e_tag("\"abc123\"") - .version_id("aws-alphanumeric-id") - .build(); - assert!( - content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_different_version)), - "matching ETags with different version IDs must still match" - ); - - let tgt_different_content = HeadObjectOutput::builder().e_tag("\"def456\"").build(); - assert!( - !content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_different_content)), - "different ETags must not match" - ); - - let src_no_etag = ObjectInfo { - etag: None, - ..Default::default() - }; - assert!( - !content_matches_by_etag(&replication_source_object(&src_no_etag), &replication_target_object(&tgt_match)), - "missing source ETag must not match" - ); - - let tgt_no_etag = HeadObjectOutput::builder().build(); - assert!( - !content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_no_etag)), - "missing target ETag must not match" - ); - } - #[test] fn test_should_count_head_proxy_failure_counts_unexpected_errors() { assert!( diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 99462f3d7..44fde7d21 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -16,8 +16,13 @@ use std::collections::HashMap; use std::sync::Arc; use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys}; +use aws_sdk_s3::operation::head_object::HeadObjectOutput; use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode}; use http::HeaderMap; +use rustfs_replication::{ + ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, + target_is_newer_than_source_null_version, +}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, @@ -35,7 +40,7 @@ pub(crate) use crate::bucket::target::BucketTargets; use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result}; -use super::replication_filemeta_boundary::ReplicationStatusType; +use super::replication_filemeta_boundary::{ReplicationAction, ReplicationStatusType, ReplicationType}; use super::replication_storage_boundary::ObjectInfo; use super::replication_tagging_boundary::ReplicationTagFilter; @@ -246,6 +251,51 @@ pub(crate) fn replication_put_object_header_size(put_options: &PutObjectOptions) .sum() } +fn replication_source_object(object_info: &ObjectInfo) -> ReplicationSourceObject<'_> { + ReplicationSourceObject { + mod_time: object_info.mod_time, + version_id: object_info.version_id.map(|version_id| version_id.to_string()), + etag: object_info.etag.as_deref(), + actual_size: object_info.get_actual_size().unwrap_or_default(), + delete_marker: object_info.delete_marker, + content_type: object_info.content_type.as_deref(), + content_encoding: object_info.content_encoding.as_deref(), + user_tags: object_info.user_tags.as_str(), + user_defined: object_info.user_defined.as_ref(), + } +} + +fn replication_target_last_modified(target: &HeadObjectOutput) -> Option { + target + .last_modified + .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) +} + +fn replication_target_object(target: &HeadObjectOutput) -> ReplicationTargetObject<'_> { + ReplicationTargetObject { + last_modified: replication_target_last_modified(target), + version_id: target.version_id.as_deref(), + etag: target.e_tag.as_deref(), + content_length: target.content_length.unwrap_or_default(), + delete_marker: target.delete_marker.unwrap_or_default(), + content_type: target.content_type.as_deref(), + metadata: target.metadata.as_ref(), + tag_count: target.tag_count.unwrap_or_default(), + } +} + +pub(crate) fn replication_action_for_target_head( + object_info: &ObjectInfo, + target: &HeadObjectOutput, + op_type: ReplicationType, +) -> ReplicationAction { + replication_action_for_target(&replication_source_object(object_info), &replication_target_object(target), op_type) +} + +pub(crate) fn replication_target_head_is_newer_null_version(object_info: &ObjectInfo, target: &HeadObjectOutput) -> bool { + target_is_newer_than_source_null_version(&replication_source_object(object_info), &replication_target_object(target)) +} + pub(crate) fn replication_delete_remove_options( delete_marker: bool, replication_mtime: Option, @@ -314,6 +364,7 @@ fn valid_sse_replication_header(key: &str) -> Option<&str> { #[cfg(test)] mod tests { use super::*; + use aws_smithy_types::DateTime; use rustfs_utils::http::{ SSEC_ALGORITHM_HEADER, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, get_header_map, }; @@ -321,6 +372,89 @@ mod tests { use time::Duration; use uuid::Uuid; + #[test] + fn replication_action_for_target_head_existing_object_source_newer_null_version_requires_replication() { + let source = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(20)), + version_id: None, + ..Default::default() + }; + let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(10)).build(); + + assert_eq!( + replication_action_for_target_head(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::All, + "a newer source null version must not be skipped during existing-object replication" + ); + } + + #[test] + fn replication_action_for_target_head_existing_object_target_newer_null_version_skips() { + let source = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + version_id: None, + ..Default::default() + }; + let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(20)).build(); + + assert_eq!( + replication_action_for_target_head(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::None, + "a newer target null-version object should not be overwritten by existing-object replication" + ); + assert!(replication_target_head_is_newer_null_version(&source, &target)); + } + + #[test] + fn replication_target_head_content_matches_compare_etag_only() { + let source = ObjectInfo { + etag: Some("\"abc123\"".to_string()), + ..Default::default() + }; + + let target_match = HeadObjectOutput::builder().e_tag("\"abc123\"").build(); + assert!( + content_matches_by_etag(&replication_source_object(&source), &replication_target_object(&target_match)), + "identical ETags must match" + ); + + let target_unquoted_match = HeadObjectOutput::builder().e_tag("abc123").build(); + assert!( + content_matches_by_etag(&replication_source_object(&source), &replication_target_object(&target_unquoted_match)), + "quoted and unquoted ETags with identical values must match" + ); + + let target_different_version = HeadObjectOutput::builder() + .e_tag("\"abc123\"") + .version_id("aws-alphanumeric-id") + .build(); + assert!( + content_matches_by_etag(&replication_source_object(&source), &replication_target_object(&target_different_version)), + "matching ETags with different version IDs must still match" + ); + + let target_different_content = HeadObjectOutput::builder().e_tag("\"def456\"").build(); + assert!( + !content_matches_by_etag(&replication_source_object(&source), &replication_target_object(&target_different_content)), + "different ETags must not match" + ); + + let source_no_etag = ObjectInfo { + etag: None, + ..Default::default() + }; + assert!( + !content_matches_by_etag(&replication_source_object(&source_no_etag), &replication_target_object(&target_match)), + "missing source ETag must not match" + ); + + let target_no_etag = HeadObjectOutput::builder().build(); + assert!( + !content_matches_by_etag(&replication_source_object(&source), &replication_target_object(&target_no_etag)), + "missing target ETag must not match" + ); + } + #[test] fn replication_remove_options_mark_replication_requests() { let mtime = OffsetDateTime::UNIX_EPOCH + Duration::seconds(10); diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 29b8bf4c3..c7dddd04b 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -15,6 +15,7 @@ pub mod config; pub mod delete; pub mod mrf; +pub mod multipart; pub mod object; pub mod operation; pub mod queue; @@ -33,6 +34,10 @@ pub use delete::{ should_retry_delete_marker_purge, }; pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file}; +pub use multipart::{ + ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange, + replication_multipart_complete_actual_size, replication_multipart_part_plan, +}; pub use object::{ ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, replication_etags_match, target_is_newer_than_source_null_version, diff --git a/crates/replication/src/multipart.rs b/crates/replication/src/multipart.rs new file mode 100644 index 000000000..e3c81a370 --- /dev/null +++ b/crates/replication/src/multipart.rs @@ -0,0 +1,222 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; +use std::fmt; + +use rustfs_utils::http::{SUFFIX_ACTUAL_SIZE, get_str}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationMultipartPartInput { + pub offset: i64, + pub part_number: usize, + pub part_size: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationMultipartPartPlan { + pub part_number: i32, + pub part_size: i64, + pub range: ReplicationMultipartRange, + pub next_offset: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicationMultipartRange { + pub start: i64, + pub end: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicationMultipartPlanError { + InvalidOffset { offset: i64 }, + InvalidPartSize { part_size: i64 }, + PartRangeOverflow { offset: i64, part_size: i64 }, + PartOffsetOverflow { offset: i64, part_size: i64 }, + PartNumberOverflow { part_number: usize }, +} + +impl fmt::Display for ReplicationMultipartPlanError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidOffset { offset } => write!(f, "invalid multipart replication part offset {offset}"), + Self::InvalidPartSize { part_size } => write!(f, "invalid multipart replication part size {part_size}"), + Self::PartRangeOverflow { offset, part_size } => { + write!(f, "multipart replication part range overflows for offset {offset} and size {part_size}") + } + Self::PartOffsetOverflow { offset, part_size } => { + write!(f, "multipart replication next offset overflows for offset {offset} and size {part_size}") + } + Self::PartNumberOverflow { part_number } => { + write!(f, "multipart replication part number {part_number} overflows i32") + } + } + } +} + +impl std::error::Error for ReplicationMultipartPlanError {} + +pub fn replication_multipart_part_plan( + input: ReplicationMultipartPartInput, +) -> Result { + if input.offset < 0 { + return Err(ReplicationMultipartPlanError::InvalidOffset { offset: input.offset }); + } + if input.part_size <= 0 { + return Err(ReplicationMultipartPlanError::InvalidPartSize { + part_size: input.part_size, + }); + } + + let part_number = i32::try_from(input.part_number).map_err(|_| ReplicationMultipartPlanError::PartNumberOverflow { + part_number: input.part_number, + })?; + let end = input + .offset + .checked_add(input.part_size - 1) + .ok_or(ReplicationMultipartPlanError::PartRangeOverflow { + offset: input.offset, + part_size: input.part_size, + })?; + let next_offset = end.checked_add(1).ok_or(ReplicationMultipartPlanError::PartOffsetOverflow { + offset: input.offset, + part_size: input.part_size, + })?; + + Ok(ReplicationMultipartPartPlan { + part_number, + part_size: input.part_size, + range: ReplicationMultipartRange { + start: input.offset, + end, + }, + next_offset, + }) +} + +pub fn replication_multipart_complete_actual_size(user_defined: &HashMap) -> String { + get_str(user_defined, SUFFIX_ACTUAL_SIZE).unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::{ + ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange, + replication_multipart_complete_actual_size, replication_multipart_part_plan, + }; + use rustfs_utils::http::{SUFFIX_ACTUAL_SIZE, insert_str}; + use std::collections::HashMap; + + #[test] + fn multipart_part_plan_builds_range_and_next_offset() { + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: 0, + part_number: 1, + part_size: 10 + }), + Ok(ReplicationMultipartPartPlan { + part_number: 1, + part_size: 10, + range: ReplicationMultipartRange { start: 0, end: 9 }, + next_offset: 10 + }) + ); + + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: 10, + part_number: 2, + part_size: 5 + }), + Ok(ReplicationMultipartPartPlan { + part_number: 2, + part_size: 5, + range: ReplicationMultipartRange { start: 10, end: 14 }, + next_offset: 15 + }) + ); + } + + #[test] + fn multipart_part_plan_rejects_invalid_offsets_and_sizes() { + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: -1, + part_number: 1, + part_size: 10 + }), + Err(ReplicationMultipartPlanError::InvalidOffset { offset: -1 }) + ); + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: 0, + part_number: 1, + part_size: 0 + }), + Err(ReplicationMultipartPlanError::InvalidPartSize { part_size: 0 }) + ); + } + + #[test] + fn multipart_part_plan_rejects_overflow() { + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: i64::MAX - 5, + part_number: 1, + part_size: 10 + }), + Err(ReplicationMultipartPlanError::PartRangeOverflow { + offset: i64::MAX - 5, + part_size: 10 + }) + ); + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: i64::MAX, + part_number: 1, + part_size: 1 + }), + Err(ReplicationMultipartPlanError::PartOffsetOverflow { + offset: i64::MAX, + part_size: 1 + }) + ); + } + + #[test] + fn multipart_part_plan_rejects_part_number_overflow() { + let overflowing_part_number = usize::MAX; + + assert_eq!( + replication_multipart_part_plan(ReplicationMultipartPartInput { + offset: 0, + part_number: overflowing_part_number, + part_size: 10 + }), + Err(ReplicationMultipartPlanError::PartNumberOverflow { + part_number: overflowing_part_number + }) + ); + } + + #[test] + fn multipart_complete_actual_size_reads_compatible_metadata() { + let mut user_defined = HashMap::new(); + insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, "123".to_string()); + + assert_eq!(replication_multipart_complete_actual_size(&user_defined), "123"); + assert!(replication_multipart_complete_actual_size(&HashMap::new()).is_empty()); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index c609a216f..a14a2e6ea 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -3288,10 +3288,14 @@ fi } ' | rg -v '^crates/ecstore/src/bucket/replication/replication_target_boundary\.rs:' || true + rg -n -U --with-filename 'rustfs_replication::\{[^;]*\bReplication(?:Source|Target)Object\b|rustfs_replication::Replication(?:Source|Target)Object\b' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' | + rg -v '^crates/ecstore/src/bucket/replication/replication_target_boundary\.rs:' || true ) >"$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" if [[ -s "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" ]]; then - report_failure "replication bucket-target access and types must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")" + report_failure "replication bucket-target access, types, and target comparison DTO adapters must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")" fi (