refactor(replication): isolate multipart and target adapters (#4211)

* refactor(replication): isolate multipart part planning

* refactor(replication): isolate target head adapters

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`. |

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<OffsetDateTime> {
-    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<S>,
     cli: Arc<TargetClient>,
@@ -2956,11 +2889,18 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(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<S: ReplicationObjectIO>(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<S: ReplicationObjectIO>(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<S: ReplicationObjectIO>(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!(

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<OffsetDateTime> {
+    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<OffsetDateTime>,
@@ -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);

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,

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<ReplicationMultipartPartPlan, ReplicationMultipartPlanError> {
+    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, String>) -> 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());
+    }
+}

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
 
 (