fix(storage): derive multipart identity from stored parts (#7305)

2026-09-07 20:46:11 +00:00 · 2026-09-06 22:04:15 +08:00
parent 8a20498705
commit 22bff27aee
12 changed files with 1269 additions and 55 deletions
@@ -21,7 +21,7 @@
 use super::common::{BoxError, OdmSourceSpec, OdmTestEnv, SeedObject};
 use crate::fake_s3_target::{BucketMode, Operation};
 use aws_sdk_s3::error::ProvideErrorMetadata;
-use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+use aws_sdk_s3::types::{BucketVersioningStatus, ObjectAttributes, VersioningConfiguration};
 use bytes::Bytes;
 use std::time::Duration;

@@ -103,11 +103,19 @@ async fn get_miss_pulls_inline_and_serves_locally_afterwards() -> TestResult {

 #[tokio::test]
 async fn get_large_object_streams_through_and_backfills_in_background() -> TestResult {
+    const PART_SIZE: usize = 5 * 1024 * 1024;
    let bucket = "odm-get-large";
-    let env = configured_env(bucket, |spec| spec.policy.inline_max_bytes = 4096).await?;
+    let env = configured_env(bucket, |spec| {
+        spec.policy.inline_max_bytes = 4096;
+        spec.policy.multipart_part_size_bytes = u64::try_from(PART_SIZE).expect("part size fits in u64");
+    })
+    .await?;
    let key = "large/archive.bin";
-    let body = payload(512 * 1024);
-    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+    let body = payload(PART_SIZE + 4096);
+    let etag = env
+        .seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())])
+        .remove(0);
+    assert_eq!(etag.len(), 32, "the source fixture has a plain MD5 ETag");

    let response = env.raw_get(bucket, key).await?;
    assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body));
@@ -125,6 +133,68 @@ async fn get_large_object_streams_through_and_backfills_in_background() -> TestR
        vec![None, None],
        "one passthrough GET plus one background pull, both unranged"
    );
+
+    let source_requests = env.source.requests().len();
+    let second_part = env.client.get_object().bucket(bucket).key(key).part_number(2).send().await?;
+    assert_eq!(second_part.content_length(), Some(4096), "the completed second part is the tail");
+    assert_eq!(
+        second_part.content_range(),
+        Some(format!("bytes {PART_SIZE}-{}/{}", body.len() - 1, body.len()).as_str()),
+        "partNumber reads the stored multipart boundary"
+    );
+    assert_eq!(
+        second_part.body.collect().await?.into_bytes(),
+        body.slice(PART_SIZE..),
+        "the local second part contains the exact source tail"
+    );
+    let third_part = env
+        .client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .part_number(3)
+        .send()
+        .await
+        .expect_err("the completed object has exactly two parts");
+    assert_eq!(third_part.code(), Some("InvalidPart"));
+
+    let mut part_marker = None;
+    for (part_number, part_size) in [(1, PART_SIZE), (2, 4096)] {
+        let attributes = env
+            .client
+            .get_object_attributes()
+            .bucket(bucket)
+            .key(key)
+            .object_attributes(ObjectAttributes::ObjectParts)
+            .object_attributes(ObjectAttributes::Etag)
+            .max_parts(1)
+            .set_part_number_marker(part_marker.clone())
+            .send()
+            .await?;
+        assert_eq!(
+            attributes.e_tag().map(|value| value.trim_matches('"')),
+            Some(etag.as_str()),
+            "multipart write-back preserves the source MD5 ETag"
+        );
+        let parts = attributes
+            .object_parts()
+            .expect("RustFS must expose the stored multipart layout");
+        assert_eq!(parts.total_parts_count(), Some(2));
+        assert_eq!(parts.max_parts(), Some(1));
+        assert_eq!(parts.is_truncated(), Some(part_number == 1));
+        assert_eq!(parts.parts().len(), 1, "RustFS returns one stored part per requested page");
+        assert_eq!(parts.parts()[0].part_number(), Some(part_number));
+        assert_eq!(parts.parts()[0].size(), Some(i64::try_from(part_size).expect("part size fits in i64")));
+        part_marker = parts.next_part_number_marker().map(str::to_owned);
+        if part_number == 1 {
+            assert_eq!(part_marker.as_deref(), Some("1"), "the next request continues after the first part");
+        }
+    }
+    assert_eq!(
+        env.source.requests().len(),
+        source_requests,
+        "local part reads must not consult the source"
+    );
    Ok(())
 }

@@ -23,16 +23,20 @@

 use super::common::{
    ALLOW_LOOPBACK_SOURCE_ENV, AdminResponse, BackfillOp, BackfillRequest, BoxError, ODM_MODULE_SWITCH_ENV, ODM_SERVER_ENV,
-    OdmEnvOptions, OdmSourceSpec, OdmTestEnv, SeedObject, start_configured_env, start_configured_env_with,
+    OdmEnvOptions, OdmSourceSpec, OdmTestEnv, SeedObject, start_configured_env, start_configured_env_with, start_source_rustfs,
 };
 use crate::common::{RustFSTestEnvironment, replication_fast_env, signed_request};
 use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, Operation};
 use crate::object_lock::common::put_object_lock_configuration;
+use crate::replication_extension_test::{
+    ReplicationTargetOptions, enable_bucket_versioning, set_replication_target_with_options,
+};
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::types::{
    BucketVersioningStatus, Event, FilterRule, FilterRuleName, NotificationConfiguration, NotificationConfigurationFilter,
-    ObjectLockRetentionMode, QueueConfiguration, S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault,
-    ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging, VersioningConfiguration,
+    ObjectAttributes, ObjectLockRetentionMode, QueueConfiguration, S3KeyFilter, ServerSideEncryption,
+    ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging,
+    VersioningConfiguration,
 };
 use bytes::Bytes;
 use local_ip_address::local_ip;
@@ -580,6 +584,130 @@ async fn test_odm_pulled_object_replicates_and_target_as_source_is_rejected() ->
        "a bucket may not migrate from its own replication target: {}",
        rejected.body
    );
+    Box::pin(assert_odm_multipart_replicates_to_rustfs(&env, bucket)).await?;
+    Ok(())
+}
+
+async fn assert_odm_multipart_replicates_to_rustfs(env: &OdmTestEnv, bucket: &str) -> TestResult {
+    const PART_SIZE: usize = 5 * 1024 * 1024;
+    let replica = start_source_rustfs().await?;
+    let replica_bucket = "odm-real-replica";
+    replica.create_test_bucket(replica_bucket).await?;
+    enable_bucket_versioning(&replica, replica_bucket).await?;
+    let arn = set_replication_target_with_options(
+        &env.rustfs,
+        bucket,
+        ReplicationTargetOptions {
+            endpoint: &replica.address,
+            access_key: &replica.access_key,
+            secret_key: &replica.secret_key,
+            target_bucket: replica_bucket,
+            secure: false,
+            skip_tls_verify: false,
+            ca_cert_pem: None,
+        },
+    )
+    .await?;
+    put_bucket_replication(&env.rustfs, bucket, &arn).await?;
+    let mut spec = env.fake_source_spec(SOURCE_BUCKET);
+    // Below the 16 MiB inline default the pull is one tee'd PUT with a single
+    // part; force the passthrough + background multipart write-back instead.
+    spec.policy.inline_max_bytes = 4096;
+    spec.policy.multipart_part_size_bytes = PART_SIZE as u64;
+    spec.policy.preserve_etag = true;
+    env.configure_and_wait(bucket, &spec).await?;
+
+    let key = "replicated/preserved-md5-multipart.bin";
+    let body = payload(PART_SIZE + 4096);
+    let source_put = env
+        .source_client()
+        .put_object()
+        .bucket(SOURCE_BUCKET)
+        .key(key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(body.clone()))
+        .send()
+        .await?;
+    let etag = source_put.e_tag().ok_or("source PUT omitted its ETag")?.trim_matches('"');
+    assert_eq!(etag.len(), 32, "the source must retain a single-PUT MD5 ETag");
+    assert!(etag.bytes().all(|byte| byte.is_ascii_hexdigit()));
+    let pulled = env.raw_get(bucket, key).await?;
+    assert_eq!(pulled.status, 200, "{}", String::from_utf8_lossy(&pulled.body));
+    assert_eq!(pulled.body, body);
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the multipart pull must persist");
+
+    let deadline = Instant::now() + SETTLE;
+    let source_head = loop {
+        let head = env.client.head_object().bucket(bucket).key(key).send().await?;
+        match head.replication_status().map(|status| status.as_str()) {
+            Some("COMPLETED") => break head,
+            Some("FAILED") => return Err("the ODM multipart copy failed replication to RustFS".into()),
+            _ => {
+                assert!(Instant::now() < deadline, "the ODM multipart copy never completed replication to RustFS");
+                tokio::time::sleep(Duration::from_millis(200)).await;
+            }
+        }
+    };
+    let version = source_head
+        .version_id()
+        .ok_or("the versioned ODM copy omitted its version id")?;
+    assert_ne!(version, "null");
+    let replica_client = replica.create_s3_client();
+    for (client, object_bucket) in [(&env.client, bucket), (&replica_client, replica_bucket)] {
+        let attributes = client
+            .get_object_attributes()
+            .bucket(object_bucket)
+            .key(key)
+            .version_id(version)
+            .object_attributes(ObjectAttributes::Etag)
+            .object_attributes(ObjectAttributes::ObjectParts)
+            .send()
+            .await?;
+        assert_eq!(attributes.e_tag().map(|value| value.trim_matches('"')), Some(etag));
+        let parts = attributes
+            .object_parts()
+            .ok_or("the local copy and replica must both expose two parts")?;
+        assert_eq!(parts.total_parts_count(), Some(2));
+        assert_eq!(
+            parts
+                .parts()
+                .iter()
+                .map(|part| (part.part_number(), part.size()))
+                .collect::<Vec<_>>(),
+            [(Some(1), Some(PART_SIZE as i64)), (Some(2), Some(4096))]
+        );
+    }
+    // REPLICA status surfaces on HEAD, like the other inbound-replica checks.
+    let replica_head = replica_client
+        .head_object()
+        .bucket(replica_bucket)
+        .key(key)
+        .version_id(version)
+        .send()
+        .await?;
+    assert_eq!(replica_head.replication_status().map(|status| status.as_str()), Some("REPLICA"));
+    let replica_get = replica_client
+        .get_object()
+        .bucket(replica_bucket)
+        .key(key)
+        .version_id(version)
+        .send()
+        .await?;
+    assert_eq!(replica_get.version_id(), Some(version));
+    assert_eq!(replica_get.body.collect().await?.into_bytes(), body);
+    let boundary = replica_client
+        .get_object()
+        .bucket(replica_bucket)
+        .key(key)
+        .version_id(version)
+        .range(format!("bytes={}-{}", PART_SIZE - 32, PART_SIZE + 31))
+        .send()
+        .await?;
+    assert_eq!(boundary.body.collect().await?.into_bytes(), body.slice(PART_SIZE - 32..PART_SIZE + 32));
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        2,
+        "one passthrough GET plus one background pull; replication and local reads must not fetch the migration source again"
+    );
    Ok(())
 }

@@ -31,17 +31,19 @@
 //! Adding a target behavior the fleet has shown: add the mode to the fake
 //! target, add a row here, and record any cell that is red before the fix.

-use crate::common::{RustFSTestEnvironment, init_logging, replication_fast_env};
-use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY};
+use crate::common::{init_logging, replication_fast_env};
+use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY};
 use crate::fake_s3_target::{FakeS3Target, Operation as FakeTargetOperation, RequestRecord};
-use crate::on_demand_migration::common::fake_source_client;
+use crate::on_demand_migration::common::{OdmEnvOptions, OdmTestEnv, fake_source_client};
 use crate::replication_extension_test::{
    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, put_bucket_replication,
    set_replication_target_with_options,
 };
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::{ByteStream, DateTime};
-use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockMode};
+use aws_sdk_s3::types::{
+    Checksum, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus, ObjectLockMode,
+};
 use bytes::Bytes;
 use std::error::Error;
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -110,16 +112,19 @@ enum ObjectShape {
    /// Two-part multipart upload with a GOVERNANCE retention period; the
    /// lock headers travel on CreateMultipartUpload, which has no body.
    LockedMultipart,
+    /// ODM stores two local parts while preserving a single-PUT source's MD5 ETag.
+    OdmPreservedMd5Multipart,
 }

 impl ObjectShape {
-    const ALL: [ObjectShape; 6] = [
+    const ALL: [ObjectShape; 7] = [
        ObjectShape::Empty,
        ObjectShape::Plain,
        ObjectShape::Retention,
        ObjectShape::LegalHold,
        ObjectShape::Multipart,
        ObjectShape::LockedMultipart,
+        ObjectShape::OdmPreservedMd5Multipart,
    ];

    fn key(self) -> &'static str {
@@ -130,6 +135,7 @@ impl ObjectShape {
            ObjectShape::LegalHold => "matrix/legal-hold.bin",
            ObjectShape::Multipart => "matrix/multipart.bin",
            ObjectShape::LockedMultipart => "matrix/locked-multipart.bin",
+            ObjectShape::OdmPreservedMd5Multipart => "matrix/odm-preserved-md5.bin",
        }
    }

@@ -139,7 +145,8 @@ impl ObjectShape {

    /// Upload the shape to the source and return the bytes the target must
    /// end up holding.
-    async fn put(self, client: &Client, bucket: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
+    async fn put(self, env: &OdmTestEnv, bucket: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
+        let client = &env.client;
        let key = self.key();
        match self {
            ObjectShape::Empty => {
@@ -190,6 +197,7 @@ impl ObjectShape {
            }
            ObjectShape::Multipart => multipart_put(client, bucket, key, 0x44, false).await,
            ObjectShape::LockedMultipart => multipart_put(client, bucket, key, 0x55, true).await,
+            ObjectShape::OdmPreservedMd5Multipart => odm_preserved_md5_multipart(env, bucket, key).await,
        }
    }
 }
@@ -270,11 +278,15 @@ async fn run_row(mode: TargetMode) -> TestResult {
    target.create_bucket_with_object_lock(target_bucket.clone());
    mode.apply(&target);

-    let mut source_env = RustFSTestEnvironment::new().await?;
    let mut env_vars = replication_fast_env();
    env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
    env_vars.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]);
-    source_env.start_rustfs_server_with_env(vec![], &env_vars).await?;
+    let env = OdmTestEnv::start_with(OdmEnvOptions {
+        env: env_vars,
+        ..OdmEnvOptions::default()
+    })
+    .await?;
+    let source_env = &env.rustfs;

    let source_bucket = format!("matrix-{}-src", mode.slug());
    let source_client = source_env.create_s3_client();
@@ -284,9 +296,9 @@ async fn run_row(mode: TargetMode) -> TestResult {
        .object_lock_enabled_for_bucket(true)
        .send()
        .await?;
-    enable_bucket_versioning(&source_env, &source_bucket).await?;
+    enable_bucket_versioning(source_env, &source_bucket).await?;
    let target_arn = set_replication_target_with_options(
-        &source_env,
+        source_env,
        &source_bucket,
        ReplicationTargetOptions {
            endpoint: &target.address(),
@@ -299,14 +311,21 @@ async fn run_row(mode: TargetMode) -> TestResult {
        },
    )
    .await?;
-    put_bucket_replication(&source_env, &source_bucket, &target_arn).await?;
+    put_bucket_replication(source_env, &source_bucket, &target_arn).await?;

    let target_client = fake_source_client(&target);
    let mut failures = Vec::new();
    for shape in ObjectShape::ALL {
        let cell = format!("{}/{:?}", mode.slug(), shape);
-        let expected_body = shape.put(&source_client, &source_bucket).await?;
+        let expected_body = shape.put(&env, &source_bucket).await?;
        let status = wait_for_terminal_replication_status(&source_client, &source_bucket, shape.key()).await?;
+        if shape == ObjectShape::OdmPreservedMd5Multipart {
+            assert_eq!(
+                env.source.count_requests(FakeTargetOperation::GetObject, shape.key()),
+                2,
+                "one passthrough GET plus one background pull; replication must read the persisted local parts"
+            );
+        }
        let journal = target.requests();
        let outcome = match expectation(mode, shape) {
            Expectation::Completed => {
@@ -379,6 +398,36 @@ async fn check_completed_cell(
    if uploads.is_empty() {
        return Err("no upload reached the target although the source reports COMPLETED".into());
    }
+    if shape == ObjectShape::OdmPreservedMd5Multipart {
+        let key_requests: Vec<_> = journal
+            .iter()
+            .filter(|record| record.key.as_deref() == Some(shape.key()))
+            .collect();
+        for operation in [
+            FakeTargetOperation::CreateMultipartUpload,
+            FakeTargetOperation::CompleteMultipartUpload,
+        ] {
+            if !key_requests.iter().any(|record| record.operation == operation) {
+                return Err(format!("preserved-MD5 multipart object did not use {operation:?}").into());
+            }
+        }
+        if key_requests
+            .iter()
+            .any(|record| record.operation == FakeTargetOperation::PutObject)
+        {
+            return Err("preserved-MD5 multipart object used a single PutObject".into());
+        }
+        let mut part_numbers: Vec<_> = key_requests
+            .iter()
+            .filter(|record| record.operation == FakeTargetOperation::UploadPart)
+            .map(|record| record.part_number)
+            .collect();
+        part_numbers.sort_unstable();
+        part_numbers.dedup();
+        if part_numbers != [Some(1), Some(2)] {
+            return Err(format!("preserved-MD5 multipart object uploaded unexpected parts: {part_numbers:?}").into());
+        }
+    }
    if let Some(framed) = uploads.iter().find(|record| record.transport.aws_chunked) {
        return Err(format!("{cell}: an upload went out aws-chunked (rustfs#6853 framing): {framed:?}").into());
    }
@@ -455,6 +504,71 @@ async fn wait_for_terminal_replication_status(
    }
 }

+async fn odm_preserved_md5_multipart(env: &OdmTestEnv, bucket: &str, key: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
+    const PART_SIZE: usize = 5 * 1024 * 1024;
+    let origin_bucket = format!("{bucket}-origin");
+    env.source.create_bucket_with_mode(&origin_bucket, BucketMode::Unversioned);
+    let mut spec = env.fake_source_spec(&origin_bucket);
+    // Below the 16 MiB inline default the pull is one tee'd PUT with a single
+    // part; force the passthrough + background multipart write-back instead.
+    spec.policy.inline_max_bytes = 4096;
+    spec.policy.multipart_part_size_bytes = PART_SIZE as u64;
+    spec.policy.preserve_etag = true;
+    env.configure_and_wait(bucket, &spec).await?;
+
+    // A normal source PUT produces the MD5 ETag; only ODM chooses the local parts.
+    let body = payload(PART_SIZE + 4096, 0x66);
+    let source_put = env
+        .source_client()
+        .put_object()
+        .bucket(&origin_bucket)
+        .key(key)
+        .body(ByteStream::from(body.clone()))
+        .send()
+        .await?;
+    let source_etag = source_put.e_tag().ok_or("source PUT omitted its ETag")?.trim_matches('"');
+    assert_eq!(source_etag.len(), 32, "source fixture must have a single-PUT MD5 ETag");
+    assert!(source_etag.bytes().all(|byte| byte.is_ascii_hexdigit()));
+
+    let pulled = env.raw_get(bucket, key).await?;
+    assert_eq!(pulled.status, 200, "{}", String::from_utf8_lossy(&pulled.body));
+    assert_eq!(pulled.body, body);
+    assert!(
+        env.wait_local_listed(bucket, key, Duration::from_secs(30)).await?,
+        "ODM must persist the object"
+    );
+    let attributes = env
+        .client
+        .get_object_attributes()
+        .bucket(bucket)
+        .key(key)
+        .object_attributes(ObjectAttributes::Etag)
+        .object_attributes(ObjectAttributes::ObjectParts)
+        .object_attributes(ObjectAttributes::Checksum)
+        .send()
+        .await?;
+    assert_eq!(attributes.e_tag().map(|etag| etag.trim_matches('"')), Some(source_etag));
+    let parts = attributes
+        .object_parts()
+        .ok_or("the ODM copy must expose its two local parts")?;
+    assert_eq!(parts.total_parts_count(), Some(2));
+    assert_eq!(
+        parts
+            .parts()
+            .iter()
+            .map(|part| (part.part_number(), part.size()))
+            .collect::<Vec<_>>(),
+        [(Some(1), Some(PART_SIZE as i64)), (Some(2), Some(4096))]
+    );
+    assert!(
+        attributes
+            .checksum()
+            .is_none_or(|checksum| checksum == &Checksum::builder().build()),
+        "multipart routing must work without an object checksum record"
+    );
+    Ok(body)
+}
+
 async fn multipart_put(
    client: &Client,
    bucket: &str,
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+#[cfg(test)]
+pub(crate) use rustfs_filemeta::ObjectPartInfo;
 pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
    REPLICATE_EXISTING, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+#[cfg(test)]
+pub(crate) use rustfs_replication::ReplicationMultipartPlanError;
 pub use rustfs_replication::{
    MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
    delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
@@ -4656,6 +4656,58 @@ where
    result
 }

+#[derive(Debug)]
+struct MultipartReplicationReadPlan {
+    part_number: i32,
+    part_size: i64,
+    range: Option<HTTPRangeSpec>,
+    next_offset: i64,
+}
+
+fn multipart_replication_read_plan(
+    object_info: &ObjectInfo,
+    obj_opts: &ObjectOptions,
+    mut input: ReplicationMultipartPartInput,
+    stored_size: usize,
+    is_last: bool,
+) -> std::io::Result<MultipartReplicationReadPlan> {
+    let empty_last_part = is_last && input.part_size == 0 && stored_size == 0;
+    // Raw reads address stored bytes. Only untransformed legacy parts may
+    // substitute their stored size for a missing logical size.
+    if obj_opts.raw_data_movement_read || (input.part_size == 0 && !object_info.is_compressed() && !object_info.is_encrypted()) {
+        input.part_size = i64::try_from(stored_size).map_err(|_| {
+            std::io::Error::new(std::io::ErrorKind::InvalidData, "multipart replication stored part size exceeds i64")
+        })?;
+    }
+    if empty_last_part {
+        if input.offset < 0 {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "empty multipart replication part has a negative offset",
+            ));
+        }
+        let part_number = i32::try_from(input.part_number)
+            .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "multipart replication part number exceeds i32"))?;
+        return Ok(MultipartReplicationReadPlan {
+            part_number,
+            part_size: 0,
+            range: None,
+            next_offset: input.offset,
+        });
+    }
+    let plan = replication_multipart_part_plan(input).map_err(std::io::Error::other)?;
+    Ok(MultipartReplicationReadPlan {
+        part_number: plan.part_number,
+        part_size: plan.part_size,
+        range: Some(HTTPRangeSpec {
+            is_suffix_length: false,
+            start: plan.range.start,
+            end: plan.range.end,
+        }),
+        next_offset: plan.next_offset,
+    })
+}
+
 async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
    ctx: MultipartReplicationContext<'_, S>,
    upload_id: &str,
@@ -4676,35 +4728,31 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(

    let mut header_size = replication_put_object_header_size(&put_opts);
    let mut offset: i64 = 0;
-    for part_info in object_info.parts.iter() {
-        // Ciphertext passthrough (raw read) ranges over the stored part
-        // bytes; decrypted reads range over the logical plaintext parts.
-        let part_size = if obj_opts.raw_data_movement_read {
-            part_info.size as i64
-        } else {
-            part_info.actual_size
-        };
-        let part_plan = replication_multipart_part_plan(ReplicationMultipartPartInput {
-            offset,
-            part_number: part_info.number,
-            part_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,
-        };
+    for (index, part_info) in object_info.parts.iter().enumerate() {
+        let part_plan = multipart_replication_read_plan(
+            object_info,
+            obj_opts,
+            ReplicationMultipartPartInput {
+                offset,
+                part_number: part_info.number,
+                part_size: part_info.actual_size,
+            },
+            part_info.size,
+            index + 1 == object_info.parts.len(),
+        )?;
        offset = part_plan.next_offset;

-        let part_reader = storage
-            .get_object_reader(src_bucket, object, Some(range_spec), HeaderMap::new(), obj_opts)
-            .await
-            .map_err(|e| std::io::Error::other(e.to_string()))?;
-
-        let part_stream = wrap_with_bandwidth_monitor_with_header(part_reader.stream, src_bucket, arn, header_size);
+        let byte_stream = if let Some(range_spec) = part_plan.range {
+            let part_reader = storage
+                .get_object_reader(src_bucket, object, Some(range_spec), HeaderMap::new(), obj_opts)
+                .await
+                .map_err(|e| std::io::Error::other(e.to_string()))?;
+            let part_stream = wrap_with_bandwidth_monitor_with_header(part_reader.stream, src_bucket, arn, header_size);
+            async_read_to_bytestream(part_stream)
+        } else {
+            ByteStream::from_static(b"")
+        };
        header_size = 0;
-        let byte_stream = async_read_to_bytestream(part_stream);

        let object_part = cli
            .put_object_part(
@@ -4760,6 +4808,173 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
 #[cfg(test)]
 mod tests {
    use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
+    use super::super::replication_object_decision_boundary::ReplicationMultipartPlanError;
+
+    #[test]
+    fn multipart_read_plan_preserves_legacy_plain_part_ranges() {
+        const MIB: usize = 1024 * 1024;
+        let object_info = ObjectInfo {
+            etag: Some("0123456789abcdef0123456789abcdef".to_string()),
+            size: 6 * 1024 * 1024,
+            ..Default::default()
+        };
+        let mut offset = 0;
+        for (part_number, stored_size, start, end) in [
+            (1, 5 * MIB, 0, 5 * 1024 * 1024 - 1),
+            (2, MIB, 5 * 1024 * 1024, 6 * 1024 * 1024 - 1),
+        ] {
+            let plan = multipart_replication_read_plan(
+                &object_info,
+                &ObjectOptions::default(),
+                ReplicationMultipartPartInput {
+                    offset,
+                    part_number,
+                    part_size: 0,
+                },
+                stored_size,
+                part_number == 2,
+            )
+            .expect("legacy plain parts must use their stored sizes");
+            assert_eq!(plan.part_number, i32::try_from(part_number).expect("part number fits"));
+            assert_eq!(plan.part_size, i64::try_from(stored_size).expect("stored size fits"));
+            let range = plan.range.expect("a nonempty part must read a range");
+            assert!(!range.is_suffix_length);
+            assert_eq!((range.start, range.end), (start, end));
+            assert_eq!(plan.next_offset, end + 1);
+            offset = plan.next_offset;
+        }
+        assert_eq!(offset, object_info.size);
+    }
+
+    #[test]
+    fn multipart_read_plan_distinguishes_transformed_and_raw_sizes() {
+        for metadata in [
+            HashMap::from([("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string())]),
+            HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]),
+        ] {
+            let object_info = ObjectInfo {
+                user_defined: Arc::new(metadata),
+                ..Default::default()
+            };
+            assert!(object_info.is_compressed() || object_info.is_encrypted());
+            for raw in [false, true] {
+                for actual_size in [-1, 0, 5] {
+                    let result = multipart_replication_read_plan(
+                        &object_info,
+                        &ObjectOptions {
+                            raw_data_movement_read: raw,
+                            ..Default::default()
+                        },
+                        ReplicationMultipartPartInput {
+                            offset: 7,
+                            part_number: 2,
+                            part_size: actual_size,
+                        },
+                        9,
+                        true,
+                    );
+                    if !raw && actual_size <= 0 {
+                        let err = result.expect_err("transformed reads cannot substitute physical bytes for unknown plaintext");
+                        assert!(matches!(
+                            err.get_ref().and_then(|err| err.downcast_ref::<ReplicationMultipartPlanError>()),
+                            Some(ReplicationMultipartPlanError::InvalidPartSize { part_size })
+                                if *part_size == actual_size
+                        ));
+                    } else {
+                        let plan = result.expect("the selected representation has a known positive size");
+                        let expected_size = if raw { 9 } else { 5 };
+                        assert_eq!(plan.part_number, 2);
+                        assert_eq!(plan.part_size, expected_size);
+                        let range = plan.range.expect("a nonempty part must read a range");
+                        assert_eq!((range.start, range.end), (7, 7 + expected_size - 1));
+                        assert_eq!(plan.next_offset, 7 + expected_size);
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn multipart_read_plan_retains_an_empty_last_part_without_advancing() {
+        for offset in [5 * 1024 * 1024, i64::MAX] {
+            for raw in [false, true] {
+                let plan = multipart_replication_read_plan(
+                    &ObjectInfo::default(),
+                    &ObjectOptions {
+                        raw_data_movement_read: raw,
+                        ..Default::default()
+                    },
+                    ReplicationMultipartPartInput {
+                        offset,
+                        part_number: 2,
+                        part_size: 0,
+                    },
+                    0,
+                    true,
+                )
+                .expect("an empty final part needs no range read");
+                assert_eq!(plan.part_number, 2);
+                assert_eq!(plan.part_size, 0);
+                assert!(plan.range.is_none());
+                assert_eq!(plan.next_offset, offset);
+            }
+        }
+    }
+
+    #[test]
+    fn multipart_read_plan_rejects_invalid_empty_parts_and_ranges() {
+        for (offset, part_number, actual_size, stored_size, is_last) in [
+            (0, 1, 0, 0, false),
+            (0, 2, -1, 0, true),
+            (0, 2, -1, 9, true),
+            (-1, 2, 0, 0, true),
+            (0, usize::try_from(i32::MAX).expect("i32 fits usize") + 1, 0, 0, true),
+            (i64::MAX, 2, 1, 1, true),
+            (i64::MAX, 2, 2, 2, true),
+        ] {
+            let err = multipart_replication_read_plan(
+                &ObjectInfo::default(),
+                &ObjectOptions::default(),
+                ReplicationMultipartPartInput {
+                    offset,
+                    part_number,
+                    part_size: actual_size,
+                },
+                stored_size,
+                is_last,
+            )
+            .expect_err("invalid part metadata must not become a successful transport plan");
+            assert!(
+                err.kind() == std::io::ErrorKind::InvalidData
+                    || err.get_ref().is_some_and(|err| { err.is::<ReplicationMultipartPlanError>() }),
+                "the failure must preserve a typed metadata or planner error: {err}"
+            );
+        }
+    }
+
+    #[cfg(target_pointer_width = "64")]
+    #[test]
+    fn multipart_read_plan_rejects_physical_size_overflow() {
+        for raw in [false, true] {
+            let err = multipart_replication_read_plan(
+                &ObjectInfo::default(),
+                &ObjectOptions {
+                    raw_data_movement_read: raw,
+                    ..Default::default()
+                },
+                ReplicationMultipartPartInput {
+                    offset: 0,
+                    part_number: 1,
+                    part_size: 0,
+                },
+                usize::MAX,
+                true,
+            )
+            .expect_err("a physical size outside the range API must be rejected before casting");
+            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+            assert_eq!(err.to_string(), "multipart replication stored part size exceeds i64");
+        }
+    }

    #[test]
    fn same_state_terminal_retry_uses_validate_only() {
@@ -6339,4 +6554,326 @@ mod tests {
            "one target's report must not silence another's"
        );
    }
+    mod multipart_transport_tests {
+        use super::super::super::replication_filemeta_boundary::ObjectPartInfo;
+        use super::super::super::replication_storage_boundary::ObjectIO as _;
+        use super::*;
+        use bytes::Bytes;
+        use http_body_util::{BodyExt, Full};
+        use std::convert::Infallible;
+
+        #[derive(Debug)]
+        struct Source {
+            body: Bytes,
+            info: ObjectInfo,
+            ranges: StdMutex<Vec<(i64, i64)>>,
+            full_reads: std::sync::atomic::AtomicUsize,
+        }
+
+        #[async_trait::async_trait]
+        impl super::super::super::replication_storage_boundary::ObjectIO for Source {
+            type Error = Error;
+            type RangeSpec = HTTPRangeSpec;
+            type HeaderMap = HeaderMap;
+            type ObjectOptions = ObjectOptions;
+            type ObjectInfo = ObjectInfo;
+            type GetObjectReader = GetObjectReader;
+            type PutObjectReader = super::super::super::replication_storage_boundary::PutObjReader;
+
+            async fn get_object_reader(
+                &self,
+                _bucket: &str,
+                _object: &str,
+                range: Option<HTTPRangeSpec>,
+                _headers: HeaderMap,
+                opts: &ObjectOptions,
+            ) -> Result<GetObjectReader> {
+                assert_eq!(
+                    opts.version_id,
+                    self.info.version_id.map(|id| id.to_string()),
+                    "every read retains the selected source version"
+                );
+                if range.is_none() {
+                    self.full_reads.fetch_add(1, Ordering::Relaxed);
+                    return Ok(GetObjectReader {
+                        stream: Box::new(std::io::Cursor::new(self.body.clone())),
+                        object_info: self.info.clone(),
+                        buffered_body: None,
+                        body_source: Default::default(),
+                    });
+                }
+                let range = range.expect("multipart transport must request an explicit nonempty range");
+                assert!(!range.is_suffix_length);
+                assert!(range.start <= range.end, "empty parts must not issue an inverted range");
+                self.ranges.lock().expect("range journal lock").push((range.start, range.end));
+                let start = usize::try_from(range.start).expect("nonnegative start");
+                let end = usize::try_from(range.end).expect("nonnegative end");
+                let body = self.body.slice(start..=end);
+                Ok(GetObjectReader {
+                    stream: Box::new(std::io::Cursor::new(body)),
+                    object_info: self.info.clone(),
+                    buffered_body: None,
+                    body_source: Default::default(),
+                })
+            }
+
+            async fn put_object(
+                &self,
+                _bucket: &str,
+                _object: &str,
+                _data: &mut Self::PutObjectReader,
+                _opts: &ObjectOptions,
+            ) -> Result<ObjectInfo> {
+                panic!("replication must not overwrite its source")
+            }
+        }
+
+        #[derive(Debug)]
+        struct RequestRecord {
+            method: http::Method,
+            query: HashMap<String, String>,
+            headers: HeaderMap,
+            body: Bytes,
+        }
+
+        #[tokio::test]
+        async fn multipart_transport_preserves_legacy_zero_actual_sizes() {
+            run_transport(4096, None).await;
+        }
+
+        #[tokio::test]
+        async fn multipart_transport_uploads_an_empty_last_part_without_reading_a_range() {
+            run_transport(0, None).await;
+        }
+
+        #[tokio::test]
+        async fn multipart_transport_preserves_transformed_unknown_nonempty_parts() {
+            for unknown_part in [(0, 0), (1, 0), (0, -1), (1, -1)] {
+                run_transport(4096, Some(unknown_part)).await;
+            }
+        }
+
+        #[tokio::test]
+        async fn multipart_transport_preserves_transformed_empty_tail() {
+            run_transport(0, Some((1, 0))).await;
+        }
+
+        async fn run_transport(tail_size: usize, unknown_part: Option<(usize, i64)>) {
+            const FIRST_SIZE: usize = 5 * 1024 * 1024;
+            let body = Bytes::from([vec![0x35; FIRST_SIZE], vec![0xa7; tail_size]].concat());
+            let etag = faster_hex::hex_string(rustfs_utils::hash::HashAlgorithm::Md5.hash_encode(&body).as_ref());
+            let source = Arc::new(Source {
+                info: ObjectInfo {
+                    size: i64::try_from(body.len() + if unknown_part.is_some() { 16 } else { 0 }).expect("stored size"),
+                    actual_size: i64::try_from(body.len()).expect("body size"),
+                    etag: Some(etag.clone()),
+                    version_id: Some(Uuid::new_v4()),
+                    user_defined: Arc::new(if unknown_part.is_some() {
+                        HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())])
+                    } else {
+                        HashMap::new()
+                    }),
+                    parts: Arc::new(vec![
+                        ObjectPartInfo {
+                            number: 1,
+                            size: FIRST_SIZE + if unknown_part.is_some() { 8 } else { 0 },
+                            actual_size: if let Some((0, size)) = unknown_part {
+                                size
+                            } else if unknown_part.is_some() || tail_size == 0 {
+                                i64::try_from(FIRST_SIZE).expect("first part size")
+                            } else {
+                                0
+                            },
+                            ..Default::default()
+                        },
+                        ObjectPartInfo {
+                            number: 2,
+                            size: tail_size + if unknown_part.is_some() { 8 } else { 0 },
+                            actual_size: if let Some((1, size)) = unknown_part {
+                                size
+                            } else if unknown_part.is_some() {
+                                i64::try_from(tail_size).expect("tail logical size")
+                            } else {
+                                0
+                            },
+                            ..Default::default()
+                        },
+                    ]),
+                    ..Default::default()
+                },
+                body: body.clone(),
+                ranges: StdMutex::new(Vec::new()),
+                full_reads: std::sync::atomic::AtomicUsize::new(0),
+            });
+            let journal = Arc::new(StdMutex::new(Vec::<RequestRecord>::new()));
+            let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
+                .await
+                .expect("bind multipart target");
+            let endpoint = format!("http://{}", listener.local_addr().expect("multipart target address"));
+            let server_journal = journal.clone();
+            let server = tokio::spawn(async move {
+                let mut connections = JoinSet::new();
+                loop {
+                    let (stream, _) = listener.accept().await.expect("accept multipart request");
+                    let journal = server_journal.clone();
+                    connections.spawn(async move {
+                            let service = hyper::service::service_fn(move |request: hyper::Request<hyper::body::Incoming>| {
+                                let journal = journal.clone();
+                                async move {
+                                    let (request, body) = request.into_parts();
+                                    let query: HashMap<String, String> = url::form_urlencoded::parse(
+                                        request.uri.query().unwrap_or_default().as_bytes(),
+                                    ).into_owned().collect();
+                                    let body = body.collect().await.expect("read complete multipart request body").to_bytes();
+                                    let response = if request.method == http::Method::POST && query.contains_key("uploads") {
+                                        "<InitiateMultipartUploadResult><Bucket>target-bucket</Bucket><Key>object</Key><UploadId>upload-1</UploadId></InitiateMultipartUploadResult>"
+                                    } else if request.method == http::Method::PUT {
+                                        ""
+                                    } else if request.method == http::Method::POST && query.contains_key("uploadId") {
+                                        "<CompleteMultipartUploadResult><Location>http://localhost/object</Location><Bucket>target-bucket</Bucket><Key>object</Key><ETag>&quot;target-2&quot;</ETag></CompleteMultipartUploadResult>"
+                                    } else if request.method == http::Method::DELETE && query.contains_key("uploadId") {
+ ""
+ } else {
+                                        panic!("unexpected multipart request: {} {}", request.method, request.uri)
+                                    };
+                                    let response_etag = if request.method == http::Method::PUT && !query.contains_key("partNumber") {
+                                        format!("\"{}\"", faster_hex::hex_string(rustfs_utils::hash::HashAlgorithm::Md5.hash_encode(&body).as_ref()))
+                                    } else {
+                                        "\"uploaded-part\"".to_string()
+                                    };
+                                    journal.lock().expect("request journal lock").push(RequestRecord {
+                                        method: request.method, query, headers: request.headers, body,
+                                    });
+                                    Ok::<_, Infallible>(hyper::Response::builder()
+                                        .header("content-type", "application/xml")
+                                        .header("etag", response_etag)
+                                        .body(Full::new(Bytes::from_static(response.as_bytes())))
+                                        .expect("multipart response"))
+                                }
+                            });
+                            hyper::server::conn::http1::Builder::new()
+                                .serve_connection(hyper_util::rt::TokioIo::new(stream), service)
+                                .await.expect("serve multipart connection");
+                        });
+                }
+            });
+            let mut target = test_target_client(endpoint);
+            let config = target
+                .client
+                .config()
+                .to_builder()
+                .request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired)
+                .force_path_style(true)
+                .build();
+            Arc::get_mut(&mut target).expect("unshared test target").client = Arc::new(aws_sdk_s3::Client::from_conf(config));
+            let (put_opts, is_multipart) = replication_put_object_options("STANDARD", &source.info).expect("replication options");
+            let opts = ObjectOptions {
+                version_id: source.info.version_id.map(|id| id.to_string()),
+                ..Default::default()
+            };
+            let reader = source
+                .get_object_reader("source", "object", None, HeaderMap::new(), &opts)
+                .await
+                .expect("open the existing full-object stream");
+            let result = tokio::time::timeout(
+                std::time::Duration::from_secs(30),
+                replicate_all_payload_to_target(
+                    ReplicateAllPayloadContext {
+                        storage: &source,
+                        tgt_client: &target,
+                        bucket: "source",
+                        object: "object",
+                        object_info: &source.info,
+                        obj_opts: &opts,
+                        arn: &target.arn,
+                        transfer_size: i64::try_from(body.len()).expect("plaintext size"),
+                        is_multipart,
+                        put_opts,
+                    },
+                    reader,
+                ),
+            )
+            .await;
+            server.abort();
+            assert!(server.await.expect_err("fixture server is stopped").is_cancelled());
+            if let Some(error) = result.expect("replication must finish") {
+                panic!("legacy parts must replicate successfully: {error}");
+            }
+            assert_eq!(
+                source.full_reads.load(Ordering::Relaxed),
+                1,
+                "reuse the initial full stream without an extra read"
+            );
+            if unknown_part.is_some() {
+                let requests = journal.lock().expect("request journal lock");
+                assert_eq!(requests.len(), 1, "unknown transformed boundaries retain one streaming PUT");
+                let request = &requests[0];
+                assert_eq!(request.method, http::Method::PUT);
+                let source_version = source.info.version_id.map(|id| id.to_string()).expect("versioned fixture");
+                assert_eq!(
+                    request.query,
+                    HashMap::from([
+                        ("x-id".to_string(), "PutObject".to_string()),
+                        ("versionId".to_string(), source_version.clone()),
+                    ]),
+                    "single PUT carries only the SDK operation query and the source versionId the target must reuse"
+                );
+                assert_eq!(request.body, body, "single PUT includes every byte of both source parts");
+                assert_eq!(
+                    request.headers.get("content-length").expect("body length"),
+                    body.len().to_string().as_str()
+                );
+                assert_eq!(
+                    rustfs_utils::http::get_header(&request.headers, rustfs_utils::http::SUFFIX_SOURCE_ETAG).as_deref(),
+                    Some(etag.as_str())
+                );
+                assert_eq!(
+                    rustfs_utils::http::get_header(&request.headers, rustfs_utils::http::SUFFIX_SOURCE_VERSION_ID)
+                        .map(|value| value.into_owned()),
+                    Some(source_version),
+                    "single PUT preserves the selected source version"
+                );
+                assert!(
+                    source.ranges.lock().expect("range journal lock").is_empty(),
+                    "unknown logical boundaries must not issue guessed ranges"
+                );
+                return;
+            }
+
+            let requests = journal.lock().expect("request journal lock");
+            assert_eq!(requests.len(), 4, "initiate, two upload parts, and complete without retries");
+            assert!(requests[0].query.contains_key("uploads"));
+            for (index, expected) in [(1, body.slice(..FIRST_SIZE)), (2, body.slice(FIRST_SIZE..))] {
+                assert_eq!(requests[index].method, http::Method::PUT);
+                assert_eq!(requests[index].query.get("partNumber"), Some(&index.to_string()));
+                assert_eq!(requests[index].body, expected, "upload part contains the exact source range");
+                assert_eq!(
+                    requests[index].headers.get("content-length").expect("part content length"),
+                    expected.len().to_string().as_str()
+                );
+            }
+            let complete = &requests[3];
+            assert_eq!(complete.method, http::Method::POST);
+            assert_eq!(
+                rustfs_utils::http::get_header(&complete.headers, rustfs_utils::http::SUFFIX_SOURCE_ETAG).as_deref(),
+                Some(etag.as_str())
+            );
+            let complete_xml = std::str::from_utf8(&complete.body).expect("complete XML");
+            assert_eq!(
+                complete_xml.matches("<Part>").count(),
+                2,
+                "the empty final part must remain in the completion list"
+            );
+            assert!(complete_xml.contains("<PartNumber>1</PartNumber>"));
+            assert!(complete_xml.contains("<PartNumber>2</PartNumber>"));
+            let mut expected_ranges = vec![(0, i64::try_from(FIRST_SIZE - 1).expect("first end"))];
+            if tail_size > 0 {
+                expected_ranges.push((
+                    i64::try_from(FIRST_SIZE).expect("tail start"),
+                    i64::try_from(body.len() - 1).expect("tail end"),
+                ));
+            }
+            assert_eq!(*source.ranges.lock().expect("range journal lock"), expected_ranges);
+        }
+    }
 }
@@ -248,7 +248,16 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
        meta.insert(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string());
    }

-    let mut is_multipart = object_info.is_multipart();
+    // Older transformed objects can have physical parts without logical part
+    // lengths. Keep their existing whole-object transport: physical sizes are
+    // not plaintext boundaries for a multipart replication read.
+    let legacy_single_put = object_info.etag.as_deref().is_none_or(|etag| etag.len() == 32);
+    let base_is_multipart = object_info.is_multipart()
+        && !(legacy_single_put
+            && object_info.parts.len() > 1
+            && (object_info.is_compressed() || object_info.is_encrypted())
+            && object_info.parts.iter().any(|part| part.actual_size <= 0));
+    let mut is_multipart = base_is_multipart;

    if let Some(checksum_data) = &object_info.checksum
        && !checksum_data.is_empty()
@@ -259,8 +268,8 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
        } else if object_info.is_encrypted() {
            // Encrypted checksums cannot be exposed as plaintext headers, and
            // decrypt_checksums reports is_multipart=false for them (a value
-            // the response path relies on). Keep the object's own multipart
-            // flag so encrypted objects stay on the multipart route.
+            // the response path relies on). Keep the transport selected from
+            // the object's layout and readable part boundaries.
        } else {
            let (checksum_meta, checksum_record_is_multipart) = object_info.decrypt_checksums(0, &HeaderMap::new())?;
            // The checksum record describes how the *checksum* is composed,
@@ -268,9 +277,9 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
            // MULTIPART flag even on a multipart upload, so trusting it here
            // routed a 768-part object through a single PutObject and the
            // target rejected the 6 GiB body with EntityTooLarge
-            // (rustfs#6825). The object's own shape is the authority: the
+            // (rustfs#6825). The usable part layout is the authority: the
            // record may only add multipart-ness, never take it away.
-            is_multipart = object_info.is_multipart() || checksum_record_is_multipart;
+            is_multipart = base_is_multipart || checksum_record_is_multipart;

            for (key, value) in checksum_meta.iter() {
                if key != AMZ_CHECKSUM_TYPE {
@@ -278,7 +287,7 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
                }
            }

-            if !object_info.is_multipart()
+            if !base_is_multipart
                && checksum_meta
                    .get(AMZ_CHECKSUM_TYPE)
                    .is_some_and(|value| value == AMZ_CHECKSUM_TYPE_FULL_OBJECT)
@@ -516,6 +525,7 @@ fn is_standard_header(key: &str) -> bool {

 #[cfg(test)]
 mod tests {
+    use super::super::replication_filemeta_boundary::ObjectPartInfo;
    use super::*;
    use aws_smithy_types::DateTime;
    use rustfs_replication::content_matches_by_etag;
@@ -550,6 +560,109 @@ mod tests {
        checksum.to_bytes(&combined)
    }

+    fn replication_route_metadata() -> [(&'static str, Arc<HashMap<String, String>>); 4] {
+        let mut compressed = HashMap::new();
+        rustfs_utils::http::insert_str(&mut compressed, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
+        [
+            ("plain", Arc::new(HashMap::new())),
+            ("compressed", Arc::new(compressed)),
+            (
+                "encrypted",
+                Arc::new(HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())])),
+            ),
+            (
+                "ssec",
+                Arc::new(HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())])),
+            ),
+        ]
+    }
+
+    fn replication_route_object(
+        etag: Option<&str>,
+        actual_sizes: [i64; 3],
+        metadata: Arc<HashMap<String, String>>,
+    ) -> ObjectInfo {
+        ObjectInfo {
+            etag: etag.map(str::to_string),
+            size: 48,
+            actual_size: 12,
+            user_defined: metadata,
+            parts: Arc::new(
+                actual_sizes
+                    .into_iter()
+                    .enumerate()
+                    .map(|(index, actual_size)| ObjectPartInfo {
+                        number: index + 1,
+                        size: 16,
+                        actual_size,
+                        ..Default::default()
+                    })
+                    .collect(),
+            ),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn legacy_transformed_single_put_parts_keep_the_previous_replication_route() {
+        let [_, (_, compressed), (_, encrypted), (_, ssec)] = replication_route_metadata();
+        let cases = [
+            (
+                "compressed middle zero",
+                compressed.clone(),
+                Some("0123456789abcdef0123456789abcdef"),
+                [4, 0, 4],
+            ),
+            ("compressed tail unknown", compressed, None, [4, 4, -1]),
+            (
+                "encrypted middle unknown",
+                encrypted,
+                Some("gggggggggggggggggggggggggggggggg"),
+                [4, -1, 4],
+            ),
+            ("ssec tail zero", ssec.clone(), None, [4, 4, 0]),
+            ("ssec middle unknown", ssec, Some("gggggggggggggggggggggggggggggggg"), [4, -1, 4]),
+        ];
+        for (name, metadata, etag, actual_sizes) in cases {
+            for checksum in [None, Some(full_object_multipart_checksum_record())] {
+                let mut object_info = replication_route_object(etag, actual_sizes, metadata.clone());
+                object_info.checksum = checksum;
+                assert!(object_info.is_multipart(), "{name}: physical parts remain visible to metadata APIs");
+                assert!(object_info.is_compressed() || object_info.is_encrypted());
+
+                let (options, is_multipart) =
+                    replication_put_object_options("STANDARD", &object_info).expect("legacy transformed put options");
+                assert!(
+                    !is_multipart,
+                    "{name}: unknown logical part sizes must preserve the old whole-object route"
+                );
+                assert_eq!(options.internal.source_etag, etag.unwrap_or_default());
+                if metadata.contains_key(SSEC_ALGORITHM_HEADER) {
+                    assert_eq!(
+                        get_header_map(&options.user_metadata, SUFFIX_REPLICATION_SSEC_CRC).is_some(),
+                        object_info.checksum.is_some(),
+                        "SSE-C checksums retain their raw passthrough transport"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn positive_part_sizes_and_legacy_multipart_etags_keep_the_replication_route() {
+        for (name, metadata) in replication_route_metadata() {
+            for (etag, actual_sizes) in [
+                ("0123456789abcdef0123456789abcdef", [4, 4, 4]),
+                ("0123456789abcdef0123456789abcdef-3", [4, 0, -1]),
+            ] {
+                let mut object_info = replication_route_object(Some(etag), actual_sizes, metadata.clone());
+                object_info.checksum = Some(full_object_multipart_checksum_record());
+                let (_, is_multipart) = replication_put_object_options("STANDARD", &object_info).expect("multipart put options");
+                assert!(is_multipart, "{name}/{etag}: usable sizes and old multipart ETags must retain MPU");
+            }
+        }
+    }
+
    #[test]
    fn multipart_object_with_full_object_checksum_keeps_the_multipart_route() {
        // rustfs#6825: a 768-part upload was replicated with a single
@@ -582,6 +695,36 @@ mod tests {
        );
    }

+    #[test]
+    fn stored_multipart_parts_keep_the_replication_route_without_a_multipart_etag() {
+        for etag in [Some("0123456789abcdef0123456789abcdef"), None] {
+            for checksum in [None, Some(full_object_multipart_checksum_record())] {
+                let object_info = ObjectInfo {
+                    etag: etag.map(str::to_string),
+                    checksum,
+                    parts: Arc::new(
+                        (1..=2)
+                            .map(|number| ObjectPartInfo {
+                                number,
+                                ..Default::default()
+                            })
+                            .collect(),
+                    ),
+                    ..Default::default()
+                };
+                let (options, is_multipart) =
+                    replication_put_object_options("STANDARD", &object_info).expect("build put options");
+
+                assert!(
+                    is_multipart,
+                    "stored parts must retain multipart routing: etag={etag:?}, checksum={:?}",
+                    object_info.checksum
+                );
+                assert_eq!(options.internal.source_etag, etag.unwrap_or_default());
+            }
+        }
+    }
+
    #[test]
    fn checksum_record_never_changes_the_transport_a_single_part_object_needs() {
        // The mirror of the rustfs#6825 guard: an object stored as one PUT
@@ -592,6 +735,10 @@ mod tests {
        let object_info = ObjectInfo {
            etag: Some("0123456789abcdef0123456789abcdef".to_string()),
            checksum: Some(checksum.to_bytes(&[])),
+            parts: Arc::new(vec![ObjectPartInfo {
+                number: 1,
+                ..Default::default()
+            }]),
            ..Default::default()
        };

@@ -628,6 +775,19 @@ mod tests {
        let (_, is_multipart) = replication_put_object_options("STANDARD", &object_info).expect("build put options");

        assert!(is_multipart, "a composite-checksum multipart object must stay on the multipart transport");
+
+        for (name, metadata) in replication_route_metadata() {
+            let mut legacy = replication_route_object(Some("0123456789abcdef0123456789abcdef"), [4, 0, 4], metadata);
+            legacy.checksum = Some(checksum.to_bytes(&combined));
+            let (_, record_is_multipart) = legacy.decrypt_checksums(0, &HeaderMap::new()).expect("decode checksum");
+            let (_, is_multipart) = replication_put_object_options("STANDARD", &legacy).expect("legacy checksum put options");
+            if legacy.is_encrypted() {
+                assert!(!is_multipart, "{name}: encrypted checksum records must not change the old transport");
+            } else {
+                assert!(record_is_multipart, "the composite checksum must carry its own multipart signal");
+                assert!(is_multipart, "{name}: a composite record can still promote the legacy route to MPU");
+            }
+        }
    }

    #[test]
@@ -2673,7 +2673,7 @@ mod tests {
            ]),
            ..Default::default()
        };
-        assert!(!object_info.is_multipart());
+        assert!(object_info.is_multipart());
        assert!(should_use_multipart_data_movement(&object_info, false));

        let single_nonstandard_part = ObjectInfo {
@@ -3050,7 +3050,7 @@ mod tests {
            ..Default::default()
        };

-        assert!(!object_info.is_multipart());
+        assert!(object_info.is_multipart());
        assert!(object_info.parts.iter().any(|part| part.checksums.is_some()));
        let opts = data_movement_put_object_opts(&object_info, 0);
        assert!(!rustfs_utils::http::contains_key_str(&opts.user_defined, SUFFIX_PART_CHECKSUMS));
@@ -2278,6 +2278,121 @@ mod tests {
        assert_eq!(read, fixture.plaintext, "SSE-C + compression full GET must reassemble all parts");
    }

+    #[tokio::test]
+    async fn multipart_empty_tail_full_reads_preserve_plaintext() {
+        let key = [0x6Eu8; 32];
+        let part_sizes = [5 * 1024 * 1024, 0];
+        let encrypted = build_legacy_ssec_multipart_fixture(key, &part_sizes).await;
+        for (kind, mut fixture, headers) in [
+            (
+                "encrypted",
+                CompressedMultipartFixture {
+                    object_info: encrypted.object_info,
+                    stored: encrypted.ciphertext,
+                    plaintext: encrypted.plaintext,
+                },
+                ssec_headers_from_key(key),
+            ),
+            ("compressed", compressed_multipart_fixture(&part_sizes).await, HeaderMap::new()),
+            (
+                "compressed and encrypted",
+                compressed_encrypted_multipart_fixture(key, &part_sizes).await,
+                ssec_headers_from_key(key),
+            ),
+        ] {
+            fixture.object_info.etag = Some(faster_hex::hex_string(Md5::digest(&fixture.plaintext).as_ref()));
+            assert_eq!(fixture.object_info.etag.as_ref().expect("source ETag").len(), 32);
+            assert_eq!(fixture.object_info.parts.len(), 2);
+            let tail = &fixture.object_info.parts[1];
+            assert_eq!(tail.actual_size, 0, "{kind}: final part has no plaintext");
+            if kind == "compressed" {
+                assert_eq!(tail.size, 0, "unpadded compression emits no bytes for an empty part");
+            } else {
+                assert!(tail.size > 0, "{kind}: the empty part still has a stored frame");
+            }
+            let stored_size = i64::try_from(fixture.stored.len()).expect("fixture size fits i64");
+            let (mut reader, offset, length) = GetObjectReader::new(
+                Box::new(Cursor::new(fixture.stored)),
+                None,
+                &fixture.object_info,
+                &ObjectOptions::default(),
+                &headers,
+            )
+            .await
+            .expect("full transformed read must include the empty tail");
+            assert_eq!((offset, length), (0, stored_size), "{kind}: full read includes all stored parts");
+            let mut body = Vec::new();
+            reader
+                .stream
+                .read_to_end(&mut body)
+                .await
+                .expect("read through the complete decoder EOF");
+            assert_eq!(body, fixture.plaintext, "{kind}: no plaintext is added or lost by the empty tail");
+        }
+    }
+
+    #[tokio::test]
+    async fn multipart_empty_tail_full_read_authenticates_v2_final_frame() {
+        let key = [0x6Eu8; 32];
+        let plaintext = legacy_fixture_part_plaintext(1, 5 * 1024 * 1024);
+        let mut ciphertext = Vec::new();
+        let mut parts = Vec::new();
+        for (number, body) in [(1, plaintext.as_slice()), (2, b"".as_slice())] {
+            let start = ciphertext.len();
+            rustfs_rio::EncryptReader::new_multipart_v2(Cursor::new(body), key, LEGACY_FIXTURE_BASE_NONCE, number)
+                .read_to_end(&mut ciphertext)
+                .await
+                .expect("encrypt a v2 fixture part with an authenticated final frame");
+            parts.push(ObjectPartInfo {
+                number,
+                size: ciphertext.len() - start,
+                actual_size: i64::try_from(body.len()).expect("fixture plaintext size fits"),
+                ..Default::default()
+            });
+        }
+        let tail_start = parts[0].size;
+        assert_eq!(parts[1].actual_size, 0);
+        assert!(parts[1].size > 8, "the empty final frame carries more than an END marker");
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "v2-empty-tail".to_string(),
+            size: i64::try_from(ciphertext.len()).expect("fixture ciphertext size fits"),
+            etag: Some(faster_hex::hex_string(Md5::digest(&plaintext).as_ref())),
+            parts: Arc::new(parts),
+            user_defined: Arc::new(legacy_ssec_multipart_metadata(key, plaintext.len())),
+            ..Default::default()
+        };
+        for corrupt_tail in [false, true] {
+            let mut stored = ciphertext.clone();
+            if corrupt_tail {
+                // The v2 header is authenticated associated data, including
+                // the header of a final frame containing zero plaintext.
+                stored[tail_start + 5] ^= 1;
+            }
+            let (mut reader, offset, length) = GetObjectReader::new(
+                Box::new(Cursor::new(stored)),
+                None,
+                &object_info,
+                &ObjectOptions::default(),
+                &ssec_headers_from_key(key),
+            )
+            .await
+            .expect("construct the full reader before consuming the final frame");
+            assert_eq!((offset, length), (0, object_info.size));
+            let result = tokio::io::copy(&mut reader.stream, &mut tokio::io::sink()).await;
+            if corrupt_tail {
+                let err = result.expect_err("EOF must authenticate the empty final frame after all plaintext is returned");
+                assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+                assert_eq!(err.to_string(), "v2 encrypted frame failed authentication");
+            } else {
+                assert_eq!(
+                    result.expect("valid empty final frame must reach EOF"),
+                    u64::try_from(plaintext.len()).expect("plaintext length fits")
+                );
+            }
+        }
+    }
+
    #[tokio::test]
    async fn compressed_encrypted_multipart_range_crosses_part_boundary() {
        let key_bytes = [0x6Eu8; 32];
@@ -3656,6 +3771,61 @@ mod tests {
        .await;
    }

+    #[tokio::test]
+    async fn multipart_full_read_preserves_legacy_zero_and_negative_part_sizes() {
+        let key = [0x77; 32];
+        let part_sizes = [5 * 1024 * 1024, 1024 * 1024];
+        let encrypted = build_legacy_ssec_multipart_fixture(key, &part_sizes).await;
+        // The encrypted case supplies the fixture key explicitly. This covers
+        // full decrypted reads, not managed-key acquisition.
+        for (kind, fixture, headers) in [
+            ("compressed", compressed_multipart_fixture(&part_sizes).await, HeaderMap::new()),
+            (
+                "encrypted with supplied key",
+                CompressedMultipartFixture {
+                    object_info: encrypted.object_info,
+                    stored: encrypted.ciphertext,
+                    plaintext: encrypted.plaintext,
+                },
+                ssec_headers_from_key(key),
+            ),
+        ] {
+            let source_etag = faster_hex::hex_string(Md5::digest(&fixture.plaintext).as_ref());
+            assert_eq!(source_etag.len(), 32);
+            assert_eq!(fixture.plaintext.len(), 6 * 1024 * 1024);
+            for part_index in 0..part_sizes.len() {
+                assert!(fixture.object_info.parts[part_index].actual_size > 0, "the selected part is nonempty");
+                for actual_size in [0, -1] {
+                    let mut object_info = fixture.object_info.clone();
+                    object_info.etag = Some(source_etag.clone());
+                    Arc::make_mut(&mut object_info.parts)[part_index].actual_size = actual_size;
+                    let (mut reader, offset, length) = GetObjectReader::new(
+                        Box::new(Cursor::new(fixture.stored.clone())),
+                        None,
+                        &object_info,
+                        &ObjectOptions::default(),
+                        &headers,
+                    )
+                    .await
+                    .expect("the authoritative total size must keep full legacy reads available");
+                    assert_eq!(offset, 0);
+                    assert_eq!(length, i64::try_from(fixture.stored.len()).expect("stored size fits"));
+                    let mut body = Vec::new();
+                    reader
+                        .stream
+                        .read_to_end(&mut body)
+                        .await
+                        .expect("full read must reach EOF despite an unspecified per-part logical size");
+                    assert_eq!(
+                        body, fixture.plaintext,
+                        "{kind}: part {part_index} with actual_size={actual_size} must not lose readable data"
+                    );
+                    assert_eq!(reader.object_info.etag.as_deref(), Some(source_etag.as_str()));
+                }
+            }
+        }
+    }
+
    /// The physical part sizes must add up to `oi.size` for a seek to be safe;
    /// inconsistent metadata must fall back to the previous full-object read
    /// instead of scheduling an erasure read past the object end.
@@ -1597,7 +1597,7 @@ impl ObjectInfo {
    }

    pub fn is_multipart(&self) -> bool {
-        self.etag.as_ref().is_some_and(|v| v.len() != 32)
+        self.parts.len() > 1 || self.etag.as_ref().is_some_and(|v| v.len() != 32)
    }

    pub fn is_encrypted(&self) -> bool {
@@ -2235,6 +2235,35 @@ mod tests {
    }
    use rustfs_filemeta::{FileInfo, FileMeta, MetaCacheEntry, TRANSITION_COMPLETE};

+    #[test]
+    fn multipart_identity_uses_stored_parts_and_preserves_the_etag_fallback() {
+        let plain_etag = "0123456789abcdef0123456789abcdef";
+        let multipart_etag = "0123456789abcdef0123456789abcdef-1";
+        for (case, part_count, etag, expected) in [
+            ("preserved source ETag", 2, Some(plain_etag), true),
+            ("missing ETag", 2, None, true),
+            ("ordinary PUT", 1, Some(plain_etag), false),
+            ("ordinary PUT without ETag", 1, None, false),
+            ("single-part MPU", 1, Some(multipart_etag), true),
+            ("legacy MPU without parts", 0, Some(multipart_etag), true),
+        ] {
+            let object = ObjectInfo {
+                etag: etag.map(str::to_string),
+                parts: Arc::new(
+                    (1..=part_count)
+                        .map(|number| ObjectPartInfo {
+                            number,
+                            ..Default::default()
+                        })
+                        .collect(),
+                ),
+                ..Default::default()
+            };
+
+            assert_eq!(object.is_multipart(), expected, "{case}");
+        }
+    }
+
    fn inline_fast_path_object(size: i64, versioned: bool) -> ObjectInfo {
        ObjectInfo {
            size,
@@ -4373,7 +4373,7 @@ mod tests {
                    }
                    retry_source_info.parts = Arc::new(retry_source_parts);
                    assert_eq!(retry_source_info.etag.as_deref(), Some(retry_object_etag.as_str()));
-                    assert!(!retry_source_info.is_multipart());
+                    assert!(retry_source_info.is_multipart());
                    assert!(retry_source_info.parts.iter().all(|part| part.checksums.is_some()));
                    assert_eq!(retry_source_info.checksum.as_deref(), Some(retry_object_checksum_bytes.as_ref()));
                    assert!(